From 83eba7027efbd2693dabacfd9eac4a3a9706bf7e Mon Sep 17 00:00:00 2001 From: ZHE YOU LIU Date: Tue, 6 Oct 2026 12:32:50 +0000 Subject: [PATCH 01/11] Record the source file of each Go SDK native Dag airflow.Dag records the file that called it, and --airflow-metadata lists it per native Dag under dag_source_files so the packer can embed the file. Task handlers add nothing. --- go-sdk/airflow/bundle.go | 25 +++++-- go-sdk/airflow/dag.go | 7 ++ go-sdk/airflow/dag_factory_test.go | 24 +++++++ go-sdk/airflow/dag_source_test.go | 65 +++++++++++++++++++ go-sdk/airflow/serve.go | 6 +- .../airflowmetadata/airflowmetadata.go | 3 + go-sdk/internal/bundle/task.go | 5 ++ go-sdk/pkg/execution/metadata.go | 5 ++ go-sdk/pkg/execution/metadata_test.go | 36 ++++++++++ 9 files changed, 171 insertions(+), 5 deletions(-) create mode 100644 go-sdk/airflow/dag_factory_test.go create mode 100644 go-sdk/airflow/dag_source_test.go diff --git a/go-sdk/airflow/bundle.go b/go-sdk/airflow/bundle.go index df10c5d5255fe..ee0c738038804 100644 --- a/go-sdk/airflow/bundle.go +++ b/go-sdk/airflow/bundle.go @@ -217,6 +217,18 @@ func (m *dagMap) add(dag *DagRef) { m.order = append(m.order, dag) } +// ListDagSourceFiles returns the file that declared each registered Dag, by dag_id. +func (m *dagMap) ListDagSourceFiles() map[string]string { + m.mu.Lock() + defer m.mu.Unlock() + + files := make(map[string]string, len(m.dags)) + for id, dag := range m.dags { + files[id] = dag.file + } + return files +} + func (m *dagMap) has(dagID string) bool { m.mu.Lock() defer m.mu.Unlock() @@ -254,18 +266,23 @@ func serializeRecovering(dag *DagRef, fileloc, relativeFileloc string) (s bundle return s } -// coordinatorSource is what Serve hands to execution.Serve: the task handlers and the Dags of one -// bundle. +// coordinatorSource is what Serve hands to execution.Serve and execution.DumpAirflowMetadata: the +// task handlers and the Dags of one bundle. type coordinatorSource struct { *taskHandlerMap dags *dagMap } var ( - _ bundle.Bundle = coordinatorSource{} - _ bundle.DagSerializer = coordinatorSource{} + _ bundle.Bundle = coordinatorSource{} + _ bundle.DagSerializer = coordinatorSource{} + _ bundle.DagSourceLister = coordinatorSource{} ) func (s coordinatorSource) SerializeDags(fileloc, relativeFileloc string) []bundle.SerializedDag { return s.dags.serialize(fileloc, relativeFileloc) } + +func (s coordinatorSource) ListDagSourceFiles() map[string]string { + return s.dags.ListDagSourceFiles() +} diff --git a/go-sdk/airflow/dag.go b/go-sdk/airflow/dag.go index 9415e514caf2c..0e4736ee5fd63 100644 --- a/go-sdk/airflow/dag.go +++ b/go-sdk/airflow/dag.go @@ -33,6 +33,9 @@ import ( // DagRef is a Dag authored in Go. [Dag] returns a new one. type DagRef struct { dagID string + // file is the source file that called Dag, as the compiler recorded it. It is empty when the + // runtime cannot report a caller. + file string // Dag and Task copy the specs they are given with copySpec, so a caller cannot change a // registered Dag through a spec it still holds. spec DagSpec @@ -71,6 +74,9 @@ type DagRef struct { // [IfRef.Then], [IfRef.Else], [SwitchRef.Case] and the methods of [TaskGroupRef] panic once the Dag // is registered. // +// The bundle embeds the source file that calls Dag, so call it from the file that declares the Dag. +// A Dag built in a factory function belongs to the file of the factory. +// // [BundleRef.Serve] sends the registered Dags to the Dag processor, but does not yet list them in // the --airflow-metadata manifest or run their tasks. // @@ -91,6 +97,7 @@ func Dag(dagID string, spec ...DagSpec) *DagRef { )) } d := &DagRef{dagID: dagID} + _, d.file, _, _ = runtime.Caller(1) if len(spec) == 1 { if err := checkDagSpec(spec[0]); err != nil { panic(fmt.Sprintf("airflow.Dag: Dag %q: %v", dagID, err)) diff --git a/go-sdk/airflow/dag_factory_test.go b/go-sdk/airflow/dag_factory_test.go new file mode 100644 index 0000000000000..ca794c5e670b1 --- /dev/null +++ b/go-sdk/airflow/dag_factory_test.go @@ -0,0 +1,24 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package airflow + +// newFactoryDag builds a Dag in a file other than the one that calls it, so a test can check +// that the Dag belongs to this file. +func newFactoryDag(dagID string) *DagRef { + return Dag(dagID) +} diff --git a/go-sdk/airflow/dag_source_test.go b/go-sdk/airflow/dag_source_test.go new file mode 100644 index 0000000000000..662a1f9b86a16 --- /dev/null +++ b/go-sdk/airflow/dag_source_test.go @@ -0,0 +1,65 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package airflow + +import ( + "bytes" + "path/filepath" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "gopkg.in/yaml.v3" + + "github.com/apache/airflow/go-sdk/internal/airflowmetadata" +) + +func TestDagRecordsCallerFile(t *testing.T) { + assert.Equal(t, "dag_source_test.go", filepath.Base(Dag("direct").file)) +} + +func TestDagFromFactoryRecordsFactoryFile(t *testing.T) { + assert.Equal(t, "dag_factory_test.go", filepath.Base(newFactoryDag("built").file)) +} + +func TestServeListsDagSourceFiles(t *testing.T) { + b := Bundle() + b.Register( + TaskHandler("py_etl", "extract", noop), + Dag("direct"), + newFactoryDag("built"), + ) + + var stdout bytes.Buffer + require.NoError(t, b.serve([]string{"--airflow-metadata"}, &stdout)) + + var got airflowmetadata.Manifest + require.NoError(t, yaml.Unmarshal(stdout.Bytes(), &got)) + require.Len(t, got.DagSourceFiles, 2) + assert.Equal(t, "dag_source_test.go", filepath.Base(got.DagSourceFiles["direct"])) + assert.Equal(t, "dag_factory_test.go", filepath.Base(got.DagSourceFiles["built"])) + assert.NotContains(t, got.DagSourceFiles, "py_etl") + assert.NotContains(t, got.Dags, "direct", "native Dags stay out of dags") +} + +func TestServeOmitsDagSourceFilesWithoutNativeDags(t *testing.T) { + var stdout bytes.Buffer + require.NoError(t, etlBundle().serve([]string{"--airflow-metadata"}, &stdout)) + + assert.NotContains(t, stdout.String(), "dag_source_files") +} diff --git a/go-sdk/airflow/serve.go b/go-sdk/airflow/serve.go index 7fafd79596529..a10fa43a9108c 100644 --- a/go-sdk/airflow/serve.go +++ b/go-sdk/airflow/serve.go @@ -140,7 +140,11 @@ func (b *BundleRef) serve(args []string, stdout io.Writer) error { if err != nil { return err } - return execution.DumpAirflowMetadata(stdout, &b.taskHandlers, format) + return execution.DumpAirflowMetadata( + stdout, + coordinatorSource{&b.taskHandlers, &b.dags}, + format, + ) case modeCoordinator: return execution.Serve(coordinatorSource{&b.taskHandlers, &b.dags}, *commAddr, *logsAddr) case modeCoordinatorUsageError: diff --git a/go-sdk/internal/airflowmetadata/airflowmetadata.go b/go-sdk/internal/airflowmetadata/airflowmetadata.go index d11412d722c44..cc0a8ac91c4fa 100644 --- a/go-sdk/internal/airflowmetadata/airflowmetadata.go +++ b/go-sdk/internal/airflowmetadata/airflowmetadata.go @@ -36,6 +36,9 @@ type Manifest struct { AirflowBundleMetadataVersion string `json:"airflow_bundle_metadata_version" yaml:"airflow_bundle_metadata_version"` SDK SDK `json:"sdk" yaml:"sdk"` Dags map[string]Dag `json:"dags" yaml:"dags"` + // DagSourceFiles maps the dag_id of each Dag authored with airflow.Dag to the source file + // that declared it, as the binary was built. Only the packer reads it, to embed the files. + DagSourceFiles map[string]string `json:"dag_source_files,omitempty" yaml:"dag_source_files,omitempty"` } // SDK identifies the SDK that produced the bundle. diff --git a/go-sdk/internal/bundle/task.go b/go-sdk/internal/bundle/task.go index c3cb8cc65570a..a8aff19e9af76 100644 --- a/go-sdk/internal/bundle/task.go +++ b/go-sdk/internal/bundle/task.go @@ -71,6 +71,11 @@ type DagSerializer interface { SerializeDags(fileloc, relativeFileloc string) []SerializedDag } +// DagSourceLister reports the source file that declared each Dag of a bundle, by dag_id. +type DagSourceLister interface { + ListDagSourceFiles() map[string]string +} + type taskFunction struct { fn reflect.Value fullName string diff --git a/go-sdk/pkg/execution/metadata.go b/go-sdk/pkg/execution/metadata.go index 05dbebd70a92b..807db34fe5566 100644 --- a/go-sdk/pkg/execution/metadata.go +++ b/go-sdk/pkg/execution/metadata.go @@ -95,6 +95,11 @@ func collectManifest(b bundle.EnumerableBundle) airflowmetadata.Manifest { dag.Tasks = append(dag.Tasks, handler.TaskID) meta.Dags[handler.DagID] = dag } + if lister, ok := b.(bundle.DagSourceLister); ok { + if files := lister.ListDagSourceFiles(); len(files) > 0 { + meta.DagSourceFiles = files + } + } return meta } diff --git a/go-sdk/pkg/execution/metadata_test.go b/go-sdk/pkg/execution/metadata_test.go index d105b24d898c3..9c0928f49b95a 100644 --- a/go-sdk/pkg/execution/metadata_test.go +++ b/go-sdk/pkg/execution/metadata_test.go @@ -26,8 +26,20 @@ import ( "gopkg.in/yaml.v3" "github.com/apache/airflow/go-sdk/internal/airflowmetadata" + "github.com/apache/airflow/go-sdk/internal/bundle" ) +type handlersOnly []bundle.TaskHandlerInfo + +func (h handlersOnly) ListTaskHandlers() []bundle.TaskHandlerInfo { return h } + +type handlersAndSources struct { + handlersOnly + files map[string]string +} + +func (h handlersAndSources) ListDagSourceFiles() map[string]string { return h.files } + func sampleManifest() airflowmetadata.Manifest { return airflowmetadata.Manifest{ // "1.0" and a task named "123" must survive a YAML round-trip as strings. @@ -117,3 +129,27 @@ func TestEncodeManifest_UnsupportedFormat(t *testing.T) { _, err := encodeManifest(sampleManifest(), MetadataFormat("xml")) require.Error(t, err) } + +func TestCollectManifest_DagSourceFiles(t *testing.T) { + handlers := handlersOnly{{DagID: "d", TaskID: "t"}} + + t.Run("bundle without a lister", func(t *testing.T) { + meta := collectManifest(handlers) + assert.Nil(t, meta.DagSourceFiles) + data, err := encodeManifest(meta, MetadataFormatYAML) + require.NoError(t, err) + assert.NotContains(t, string(data), "dag_source_files") + }) + + t.Run("lister with files", func(t *testing.T) { + files := map[string]string{"native": "/src/native.go"} + meta := collectManifest(handlersAndSources{handlers, files}) + assert.Equal(t, files, meta.DagSourceFiles) + assert.Equal(t, []string{"t"}, meta.Dags["d"].Tasks) + }) + + t.Run("lister without files", func(t *testing.T) { + meta := collectManifest(handlersAndSources{handlers, nil}) + assert.Nil(t, meta.DagSourceFiles) + }) +} From d49fbc9e03a91d9eec72d165e0c3521cf437892f Mon Sep 17 00:00:00 2001 From: ZHE YOU LIU Date: Tue, 6 Oct 2026 12:35:48 +0000 Subject: [PATCH 02/11] Embed each native Dag's source file in Go bundles The packer reads dag_source_files from the binary's manifest and embeds the entrypoint plus the file of every native Dag once, in a stable order. The manifest drops source and gains entrypoint_path, dag_source_paths and a sources index. A Dag whose file is outside the module maps to the entrypoint with a warning. --- go-sdk/cmd/airflow-go-pack/main.go | 9 +- go-sdk/cmd/airflow-go-pack/pack.go | 107 +++-- .../airflow-go-pack/pack_integration_test.go | 9 +- go-sdk/cmd/airflow-go-pack/pack_test.go | 61 ++- go-sdk/cmd/airflow-go-pack/sources.go | 255 +++++++++++ go-sdk/cmd/airflow-go-pack/sources_test.go | 402 ++++++++++++++++++ .../testdata/handlersonly/main.go | 36 ++ .../testdata/multidag/factory/factory.go | 29 ++ .../airflow-go-pack/testdata/multidag/main.go | 49 +++ .../testdata/multidag/reports/reports.go | 29 ++ 10 files changed, 948 insertions(+), 38 deletions(-) create mode 100644 go-sdk/cmd/airflow-go-pack/sources.go create mode 100644 go-sdk/cmd/airflow-go-pack/sources_test.go create mode 100644 go-sdk/cmd/airflow-go-pack/testdata/handlersonly/main.go create mode 100644 go-sdk/cmd/airflow-go-pack/testdata/multidag/factory/factory.go create mode 100644 go-sdk/cmd/airflow-go-pack/testdata/multidag/main.go create mode 100644 go-sdk/cmd/airflow-go-pack/testdata/multidag/reports/reports.go diff --git a/go-sdk/cmd/airflow-go-pack/main.go b/go-sdk/cmd/airflow-go-pack/main.go index cc04b315ac42e..2f30ea5acabcd 100644 --- a/go-sdk/cmd/airflow-go-pack/main.go +++ b/go-sdk/cmd/airflow-go-pack/main.go @@ -51,10 +51,11 @@ func newRootCmd() *cobra.Command { Use: "airflow-go-pack [package]", Short: "Build a self-contained Airflow bundle from a Go package", Long: `airflow-go-pack builds a Go bundle binary, queries it for its DAG/task -identity via --airflow-metadata, and appends the source plus an +identity via --airflow-metadata, and appends the source files plus an airflow-metadata.yaml manifest plus an AFBNDL01 trailer to the -executable. The result is a single self-contained file that drops into -[executable] bundles_folder. +executable. The source files are the entrypoint (the file with func main) +and the file that declares each Dag built with airflow.Dag. The result is a +single self-contained file that drops into [executable] bundles_folder. By default the packer builds the package in the current directory. Pass a different package as the positional argument; pass extra go build @@ -121,7 +122,7 @@ Examples: root.Flags().StringVar(&opts.source, "source", "", - "path to the DAG source file (defaults to the file in the target package containing func main)") + "path to the entrypoint source file (defaults to the file in the target package containing func main)") root.Flags().StringVar(&opts.executable, "executable", "", "pack a pre-built executable instead of running go build") diff --git a/go-sdk/cmd/airflow-go-pack/pack.go b/go-sdk/cmd/airflow-go-pack/pack.go index f4be6fa8c4578..8d96330b864d2 100644 --- a/go-sdk/cmd/airflow-go-pack/pack.go +++ b/go-sdk/cmd/airflow-go-pack/pack.go @@ -30,6 +30,7 @@ import ( "path/filepath" "runtime" "sort" + "strconv" "strings" "gopkg.in/yaml.v3" @@ -41,7 +42,7 @@ import ( // packOptions are the flags accepted by the root pack command. type packOptions struct { pkg string // target package (default ".") - source string // override the auto-detected DAG source file + source string // override the auto-detected entrypoint source file executable string // pack a pre-built binary instead of building output string // override the default output path airflowMetadata string // path to a pre-captured --airflow-metadata manifest (JSON or YAML) @@ -62,13 +63,13 @@ func runPack(stdout, stderr io.Writer, opts *packOptions) error { ) } - // Resolve the DAG source file for both modes up front. --executable requires + // Resolve the entrypoint source file for both modes up front. --executable requires // it explicitly; the build path falls back to discovery. sourcePath := opts.source if opts.executable != "" { if sourcePath == "" { return fmt.Errorf( - "--executable requires --source: cannot infer the DAG source for a pre-built binary", + "--executable requires --source: cannot infer the entrypoint source for a pre-built binary", ) } } else if sourcePath == "" { @@ -80,7 +81,7 @@ func runPack(stdout, stderr io.Writer, opts *packOptions) error { // when --source was not supplied, so an explicit --source always wins. discovered, err := discoverMainSource(opts.pkg) if err != nil { - return fmt.Errorf("locating DAG source file: %w", err) + return fmt.Errorf("locating the entrypoint source file: %w", err) } sourcePath = discovered } @@ -155,7 +156,7 @@ func runPack(stdout, stderr io.Writer, opts *packOptions) error { return fmt.Errorf("executable %s: %w", execPath, err) } - if err := rejectOutputAlias(output, execPath, sourcePath, opts.airflowMetadata); err != nil { + if err := rejectOutputAlias(output, execPath, []string{sourcePath}, opts.airflowMetadata); err != nil { return err } @@ -163,7 +164,7 @@ func runPack(stdout, stderr io.Writer, opts *packOptions) error { if err != nil { return err } - if len(meta.Dags) == 0 { + if len(meta.Dags) == 0 && len(meta.DagSourceFiles) == 0 { return fmt.Errorf("bundle exposes no dags: nothing to pack") } for dagID, dag := range meta.Dags { @@ -173,20 +174,27 @@ func runPack(stdout, stderr io.Writer, opts *packOptions) error { } warnOnSuspiciousIDs(stderr, meta) - manifest, err := renderManifest(meta, filepath.Base(sourcePath)) + mod, err := moduleOf(opts, sourcePath) if err != nil { - return fmt.Errorf("rendering manifest: %w", err) + return err + } + layout, region, err := layoutSources(stderr, meta, sourcePath, mod) + if err != nil { + return err + } + if err := rejectOutputAlias(output, execPath, layout.diskPaths(), opts.airflowMetadata); err != nil { + return err } - sourceBytes, err := os.ReadFile(sourcePath) + manifest, err := renderManifest(meta, layout) if err != nil { - return fmt.Errorf("reading source file: %w", err) + return fmt.Errorf("rendering manifest: %w", err) } // Assemble the bundle through a temp file and atomically move it into // place: we never mutate the build artefact or the user-supplied // --executable, and a failed pack never leaves a truncated or half-written // file at output. - if err := writeBundle(execPath, output, sourceBytes, manifest); err != nil { + if err := writeBundle(execPath, output, region, manifest); err != nil { return err } @@ -195,6 +203,17 @@ func runPack(stdout, stderr io.Writer, opts *packOptions) error { return nil } +// moduleOf finds the module that owns the entrypoint. The build path asks the go tool, which +// knows about workspaces and vendoring. --executable has no package to ask about, so it reads +// the nearest go.mod above the source file. +func moduleOf(opts *packOptions, sourcePath string) (goModule, error) { + dir := filepath.Dir(sourcePath) + if opts.executable != "" { + return findModule(dir) + } + return listModule(dir) +} + // defaultOutputPath derives the default bundle output path from the directory // that owns the DAG source file. That directory is the bundle's main package, // so its base name is what `go build` itself would name the binary. On Windows @@ -463,11 +482,10 @@ func runIntrospect(execPath string, flag string) ([]byte, error) { } // renderManifest serialises the airflow-metadata manifest as deterministic, -// sorted-key YAML matching airflow-metadata.schema.json. It injects the schema's -// source field (the filename the manifest is built from), which the producer's -// Manifest omits because only the packer knows it; every other field is copied -// from the introspected manifest verbatim. -func renderManifest(meta airflowmetadata.Manifest, sourceName string) ([]byte, error) { +// sorted-key YAML matching airflow-metadata.schema.json. It adds the source layout +// (entrypoint_path, dag_source_paths and sources), which only the packer knows; every +// other field is copied from the introspected manifest verbatim. +func renderManifest(meta airflowmetadata.Manifest, layout sourceLayout) ([]byte, error) { version := meta.AirflowBundleMetadataVersion if version == "" { version = airflowmetadata.FormatVersion @@ -498,6 +516,33 @@ func renderManifest(meta airflowmetadata.Manifest, sourceName string) ([]byte, e ) } + dagPaths := make([]string, 0, len(layout.dagPaths)) + for id := range layout.dagPaths { + dagPaths = append(dagPaths, id) + } + sort.Strings(dagPaths) + dagPathsNode := &yaml.Node{Kind: yaml.MappingNode} + for _, id := range dagPaths { + dagPathsNode.Content = append( + dagPathsNode.Content, + scalar(id), + quotedScalar(layout.dagPaths[id]), + ) + } + + sourcesNode := &yaml.Node{Kind: yaml.SequenceNode} + for _, f := range layout.files { + sourcesNode.Content = append(sourcesNode.Content, &yaml.Node{ + Kind: yaml.MappingNode, + Content: []*yaml.Node{ + scalar("path"), quotedScalar(f.path), + scalar("offset"), intScalar(f.offset), + scalar("length"), intScalar(f.length), + scalar("sha256"), quotedScalar(f.sha256), + }, + }) + } + root := &yaml.Node{Kind: yaml.DocumentNode} manifest := &yaml.Node{ Kind: yaml.MappingNode, @@ -513,7 +558,9 @@ func renderManifest(meta airflowmetadata.Manifest, sourceName string) ([]byte, e quotedScalar(meta.SDK.SupervisorSchemaVersion), }, }, - scalar("source"), quotedScalar(sourceName), + scalar("entrypoint_path"), quotedScalar(layout.entrypoint), + scalar("dag_source_paths"), dagPathsNode, + scalar("sources"), sourcesNode, scalar("dags"), dagsNode, }, } @@ -537,8 +584,12 @@ func scalar(value string) *yaml.Node { return &yaml.Node{Kind: yaml.ScalarNode, Value: value} } +func intScalar(value int) *yaml.Node { + return &yaml.Node{Kind: yaml.ScalarNode, Tag: "!!int", Value: strconv.Itoa(value)} +} + // quotedScalar emits a double-quoted node. Data-bearing string *values* — task -// IDs, the source filename, and the SDK fields — go through this so a value +// IDs, the source paths, and the SDK fields — go through this so a value // that looks like a number, bool, or date (e.g. a task named "123" or "true") // round-trips as a string rather than being retyped by the YAML parser. func quotedScalar(value string) *yaml.Node { @@ -546,19 +597,21 @@ func quotedScalar(value string) *yaml.Node { } // rejectOutputAlias fails if output resolves to the same file as any pack -// input: the executable, the source, or a supplied --airflow-metadata file. +// input: the executable, a source file, or a supplied --airflow-metadata file. // Packing copies the executable to output with O_TRUNC and renames it into // place, so an aliased output would clobber the input. metadataPath is empty // when --airflow-metadata is not used and is skipped in that case. -func rejectOutputAlias(output, execPath, sourcePath, metadataPath string) error { - for _, in := range []struct { +func rejectOutputAlias(output, execPath string, sourcePaths []string, metadataPath string) error { + type input struct { path string kind string - }{ - {execPath, "executable"}, - {sourcePath, "source"}, - {metadataPath, "--airflow-metadata file"}, - } { + } + inputs := []input{{execPath, "executable"}} + for _, p := range sourcePaths { + inputs = append(inputs, input{p, "source"}) + } + inputs = append(inputs, input{metadataPath, "--airflow-metadata file"}) + for _, in := range inputs { if in.path == "" { continue } @@ -612,7 +665,7 @@ func sameFile(a, b string) (bool, error) { } // writeBundle assembles the bundle at output by copying the executable to a -// temporary file in output's directory, appending the source+manifest footer +// temporary file in output's directory, appending the source region and manifest footer // to that copy, then atomically renaming it into place. Writing through a // temp file keeps a failed pack from leaving a truncated or half-written // artefact at output, and guarantees the file being copied is never the same diff --git a/go-sdk/cmd/airflow-go-pack/pack_integration_test.go b/go-sdk/cmd/airflow-go-pack/pack_integration_test.go index 7f5c6b4a37300..f4ab7e4ac7b6d 100644 --- a/go-sdk/cmd/airflow-go-pack/pack_integration_test.go +++ b/go-sdk/cmd/airflow-go-pack/pack_integration_test.go @@ -26,6 +26,7 @@ import ( "path/filepath" "regexp" "runtime" + "strconv" "testing" "github.com/stretchr/testify/assert" @@ -144,7 +145,13 @@ sdk: language: "go" version: "` + sdkVersion + `" supervisor_schema_version: "` + execution.SupervisorSchemaVersion + `" -source: "main.go" +entrypoint_path: "example/bundle/main.go" +dag_source_paths: {} +sources: + - path: "example/bundle/main.go" + offset: 0 + length: ` + strconv.Itoa(len(srcBytes)) + ` + sha256: "` + sha256Hex(srcBytes) + `" dags: concurrent_xcom_dag: tasks: diff --git a/go-sdk/cmd/airflow-go-pack/pack_test.go b/go-sdk/cmd/airflow-go-pack/pack_test.go index 999d5c0f01771..f90ab92b1480a 100644 --- a/go-sdk/cmd/airflow-go-pack/pack_test.go +++ b/go-sdk/cmd/airflow-go-pack/pack_test.go @@ -23,6 +23,7 @@ import ( "os" "path/filepath" "runtime" + "strconv" "testing" "github.com/spf13/cobra" @@ -33,6 +34,15 @@ import ( "github.com/apache/airflow/go-sdk/internal/bundlefooter" ) +var testLayout = sourceLayout{ + entrypoint: "cmd/bundle/main.go", + dagPaths: map[string]string{"zeta_dag": "cmd/bundle/main.go", "alpha_dag": "dags/alpha.go"}, + files: []sourceFile{ + {path: "cmd/bundle/main.go", offset: 0, length: 10, sha256: "aa"}, + {path: "dags/alpha.go", offset: 10, length: 20, sha256: "bb"}, + }, +} + func TestRenderManifest_DeterministicDagOrdering(t *testing.T) { meta := airflowmetadata.Manifest{ AirflowBundleMetadataVersion: "1.0", @@ -47,9 +57,9 @@ func TestRenderManifest_DeterministicDagOrdering(t *testing.T) { }, } - got1, err := renderManifest(meta, "main.go") + got1, err := renderManifest(meta, testLayout) require.NoError(t, err) - got2, err := renderManifest(meta, "main.go") + got2, err := renderManifest(meta, testLayout) require.NoError(t, err) assert.Equal(t, got1, got2, "manifest should be byte-identical for identical input") @@ -59,7 +69,19 @@ sdk: language: "go" version: "0.1.0" supervisor_schema_version: "2026-06-16" -source: "main.go" +entrypoint_path: "cmd/bundle/main.go" +dag_source_paths: + alpha_dag: "dags/alpha.go" + zeta_dag: "cmd/bundle/main.go" +sources: + - path: "cmd/bundle/main.go" + offset: 0 + length: 10 + sha256: "aa" + - path: "dags/alpha.go" + offset: 10 + length: 20 + sha256: "bb" dags: alpha_dag: tasks: @@ -72,7 +94,7 @@ dags: assert.Equal(t, expected, string(got1)) } -// Values (task IDs, source, SDK fields) are quoted so a scalar-looking value +// Values (task IDs, source paths, SDK fields) are quoted so a scalar-looking value // stays a string; Dag ID keys stay plain scalars. func TestRenderManifest_QuotesValuesNotKeys(t *testing.T) { meta := airflowmetadata.Manifest{ @@ -87,7 +109,7 @@ func TestRenderManifest_QuotesValuesNotKeys(t *testing.T) { }, } - got, err := renderManifest(meta, "main.go") + got, err := renderManifest(meta, testLayout) require.NoError(t, err) // Task values that look like scalars are quoted. @@ -108,7 +130,7 @@ func TestRenderManifest_EmptyDags(t *testing.T) { }, Dags: map[string]airflowmetadata.Dag{}, } - got, err := renderManifest(meta, "main.go") + got, err := renderManifest(meta, testLayout) require.NoError(t, err) assert.Contains(t, string(got), "dags: {}") } @@ -377,6 +399,33 @@ func TestRunPack_UsesMetadataFile(t *testing.T) { assert.Equal(t, exeBytes, binaryRegion, "the supplied --executable must be packed verbatim") } +func TestRunPack_PacksABundleWithOnlyNativeDags(t *testing.T) { + dir := t.TempDir() + exe := filepath.Join(dir, "native") + require.NoError(t, os.WriteFile(exe, []byte("native-binary-bytes"), 0o755)) + source := filepath.Join(dir, "main.go") + require.NoError(t, os.WriteFile(source, []byte("package main\nfunc main() {}\n"), 0o644)) + meta := filepath.Join(dir, "airflow-metadata.json") + require.NoError(t, os.WriteFile(meta, []byte( + `{"airflow_bundle_metadata_version":"1.0",`+ + `"sdk":{"language":"go","version":"0.1.0","supervisor_schema_version":"2026-06-16"},`+ + `"dags":{},"dag_source_files":{"native_dag":`+strconv.Quote(source)+`}}`, + ), 0o644)) + out := filepath.Join(dir, "bundle") + + err := runPack(io.Discard, io.Discard, &packOptions{ + executable: exe, + source: source, + airflowMetadata: meta, + output: out, + }) + require.NoError(t, err) + + _, gotMeta, err := bundlefooter.Read(out) + require.NoError(t, err) + assert.Contains(t, string(gotMeta), `native_dag: "main.go"`) +} + // --airflow-metadata also accepts a YAML manifest, not only the JSON the // binary prints. func TestRunPack_AcceptsYAMLMetadataFile(t *testing.T) { diff --git a/go-sdk/cmd/airflow-go-pack/sources.go b/go-sdk/cmd/airflow-go-pack/sources.go new file mode 100644 index 0000000000000..a49746a2b6f8a --- /dev/null +++ b/go-sdk/cmd/airflow-go-pack/sources.go @@ -0,0 +1,255 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package main + +import ( + "bytes" + "crypto/sha256" + "encoding/hex" + "fmt" + "io" + "os" + "os/exec" + "path" + "path/filepath" + "sort" + "strings" + + "github.com/apache/airflow/go-sdk/internal/airflowmetadata" + "github.com/apache/airflow/go-sdk/internal/bundlefooter" +) + +// goModule is the module that owns the bundle's entrypoint. +type goModule struct { + dir string // module root on disk + path string // module path from go.mod; empty when no go.mod was found +} + +// sourceFile is one file embedded in the source region. +type sourceFile struct { + path string // slash path relative to the module root + disk string // where the packer read the file from + offset int + length int + sha256 string // lowercase hex +} + +// sourceLayout describes the source region: which file is the entrypoint, which file each +// native Dag came from, and where every file sits in the region. +type sourceLayout struct { + entrypoint string + dagPaths map[string]string + files []sourceFile +} + +// diskPaths lists the files the packer read, so the caller can keep the output away from them. +func (l sourceLayout) diskPaths() []string { + paths := make([]string, len(l.files)) + for i, f := range l.files { + paths[i] = f.disk + } + return paths +} + +// listModule asks the go tool for the module that owns dir. In a workspace it lists every +// workspace module, so it keeps the innermost one that contains dir. +func listModule(dir string) (goModule, error) { + cmd := exec.Command("go", "list", "-m", "-f", "{{.Dir}}\n{{.Path}}") + cmd.Dir = dir + var stdout, stderr bytes.Buffer + cmd.Stdout = &stdout + cmd.Stderr = &stderr + if err := cmd.Run(); err != nil { + return goModule{}, fmt.Errorf( + "go list -m in %s: %w: %s", dir, err, strings.TrimSpace(stderr.String()), + ) + } + absDir, err := filepath.Abs(dir) + if err != nil { + return goModule{}, err + } + lines := splitNonEmpty(stdout.String()) + var best goModule + for i := 0; i+1 < len(lines); i += 2 { + if _, ok := relativeTo(lines[i], absDir); ok && len(lines[i]) > len(best.dir) { + best = goModule{dir: lines[i], path: lines[i+1]} + } + } + if best.dir == "" { + return goModule{}, fmt.Errorf("go list -m in %s: no module contains the directory", dir) + } + return best, nil +} + +// findModule walks up from dir to the nearest go.mod and reads its module path. It is for +// --executable, where there is no package to hand to the go tool. Without a go.mod the module +// is dir itself with no path, which still resolves absolute paths under it. +func findModule(dir string) (goModule, error) { + abs, err := filepath.Abs(dir) + if err != nil { + return goModule{}, err + } + for cur := abs; ; cur = filepath.Dir(cur) { + data, err := os.ReadFile(filepath.Join(cur, "go.mod")) + if err == nil { + return goModule{dir: cur, path: modulePath(data)}, nil + } + if !os.IsNotExist(err) { + return goModule{}, err + } + if filepath.Dir(cur) == cur { + return goModule{dir: abs}, nil + } + } +} + +// modulePath returns the path on the module line of a go.mod file, or "" if there is none. +func modulePath(goMod []byte) string { + for line := range strings.SplitSeq(string(goMod), "\n") { + line, _, _ = strings.Cut(line, "//") + fields := strings.Fields(line) + if len(fields) == 2 && fields[0] == "module" { + return strings.Trim(fields[1], "\"`") + } + } + return "" +} + +// relativeTo returns the slash path of file relative to dir. It reports false if file is not +// inside dir. +func relativeTo(dir, file string) (string, bool) { + rel, err := filepath.Rel(dir, file) + if err != nil { + return "", false + } + rel = filepath.ToSlash(rel) + if rel == "." || rel == ".." || strings.HasPrefix(rel, "../") { + return "", false + } + return rel, true +} + +// moduleRelPath turns a source path reported by the bundle binary into a slash path relative +// to the module root. The binary reports an absolute build path, or "/" +// when it was built with -trimpath. It reports false for a file that is not a regular file +// inside the module, such as one from the module cache or another module. +func moduleRelPath(reported string, mod goModule) (string, bool) { + var rel string + switch { + case reported == "": + return "", false + case filepath.IsAbs(filepath.FromSlash(reported)): + var ok bool + if rel, ok = relativeTo(mod.dir, filepath.FromSlash(reported)); !ok { + return "", false + } + case mod.path != "" && strings.HasPrefix(reported, mod.path+"/"): + rel = path.Clean(strings.TrimPrefix(reported, mod.path+"/")) + if rel == ".." || strings.HasPrefix(rel, "../") { + return "", false + } + default: + return "", false + } + info, err := os.Stat(filepath.Join(mod.dir, filepath.FromSlash(rel))) + if err != nil || !info.Mode().IsRegular() { + return "", false + } + return rel, true +} + +// layoutSources decides which files the bundle embeds and what each native Dag maps to. The +// entrypoint always comes first, and every other file follows in sorted order, so identical +// inputs give an identical region. A Dag whose file lies outside the module maps to the +// entrypoint, with a warning. It returns the layout and the concatenated region. +func layoutSources( + stderr io.Writer, + meta airflowmetadata.Manifest, + entrypoint string, + mod goModule, +) (sourceLayout, []byte, error) { + absEntry, err := filepath.Abs(entrypoint) + if err != nil { + return sourceLayout{}, nil, err + } + entryPath, ok := relativeTo(mod.dir, absEntry) + if !ok { + entryPath = filepath.Base(absEntry) + } + + layout := sourceLayout{ + entrypoint: entryPath, + dagPaths: make(map[string]string, len(meta.DagSourceFiles)), + } + disk := map[string]string{entryPath: absEntry} + + dagIDs := make([]string, 0, len(meta.DagSourceFiles)) + for id := range meta.DagSourceFiles { + dagIDs = append(dagIDs, id) + } + sort.Strings(dagIDs) + for _, id := range dagIDs { + reported := meta.DagSourceFiles[id] + rel, ok := moduleRelPath(reported, mod) + if !ok { + fmt.Fprintf(stderr, + "warning: source file %q of dag %q is outside module %s; "+ + "embedding the entrypoint %s for it instead\n", + reported, id, mod.dir, entryPath) + layout.dagPaths[id] = entryPath + continue + } + layout.dagPaths[id] = rel + if _, seen := disk[rel]; !seen { + disk[rel] = filepath.Join(mod.dir, filepath.FromSlash(rel)) + } + } + + paths := make([]string, 0, len(disk)) + for p := range disk { + if p != entryPath { + paths = append(paths, p) + } + } + sort.Strings(paths) + paths = append([]string{entryPath}, paths...) + + var region bytes.Buffer + for _, p := range paths { + data, err := os.ReadFile(disk[p]) + if err != nil { + return sourceLayout{}, nil, fmt.Errorf("reading source file: %w", err) + } + sum := sha256.Sum256(data) + layout.files = append(layout.files, sourceFile{ + path: p, + disk: disk[p], + offset: region.Len(), + length: len(data), + sha256: hex.EncodeToString(sum[:]), + }) + region.Write(data) + if int64(region.Len()) > bundlefooter.MaxRegionSize { + return sourceLayout{}, nil, fmt.Errorf( + "source region too large after embedding %s (max %d bytes)", + p, int64(bundlefooter.MaxRegionSize), + ) + } + } + return layout, region.Bytes(), nil +} diff --git a/go-sdk/cmd/airflow-go-pack/sources_test.go b/go-sdk/cmd/airflow-go-pack/sources_test.go new file mode 100644 index 0000000000000..96a87c423cb04 --- /dev/null +++ b/go-sdk/cmd/airflow-go-pack/sources_test.go @@ -0,0 +1,402 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +package main + +import ( + "bytes" + "crypto/sha256" + "encoding/hex" + "os" + "os/exec" + "path/filepath" + "runtime" + "testing" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "gopkg.in/yaml.v3" + + "github.com/apache/airflow/go-sdk/internal/airflowmetadata" + "github.com/apache/airflow/go-sdk/internal/bundlefooter" +) + +func sha256Hex(b []byte) string { + sum := sha256.Sum256(b) + return hex.EncodeToString(sum[:]) +} + +// packedManifest is the part of a packed manifest that the source layout adds. +type packedManifest struct { + EntrypointPath string `yaml:"entrypoint_path"` + DagSourcePaths map[string]string `yaml:"dag_source_paths"` + Sources []struct { + Path string `yaml:"path"` + Offset int `yaml:"offset"` + Length int `yaml:"length"` + SHA256 string `yaml:"sha256"` + } `yaml:"sources"` +} + +func readPacked(t *testing.T, bundle string) (packedManifest, []byte, []byte) { + t.Helper() + region, metadata, err := bundlefooter.Read(bundle) + require.NoError(t, err) + var m packedManifest + require.NoError(t, yaml.Unmarshal(metadata, &m)) + return m, region, metadata +} + +// writeModule lays out files under a new directory with a go.mod, and returns the directory. +func writeModule(t *testing.T, modulePath string, files map[string]string) string { + t.Helper() + dir := t.TempDir() + require.NoError(t, os.WriteFile(filepath.Join(dir, "go.mod"), + []byte("module "+modulePath+"\n\ngo 1.24\n"), 0o644)) + for rel, content := range files { + full := filepath.Join(dir, filepath.FromSlash(rel)) + require.NoError(t, os.MkdirAll(filepath.Dir(full), 0o755)) + require.NoError(t, os.WriteFile(full, []byte(content), 0o644)) + } + return dir +} + +func TestModulePath(t *testing.T) { + for _, tc := range []struct { + name string + mod string + want string + }{ + {name: "plain", mod: "module example.com/app\n\ngo 1.24\n", want: "example.com/app"}, + {name: "quoted", mod: "module \"example.com/app\"\n", want: "example.com/app"}, + {name: "comment", mod: "// header\nmodule example.com/app // trailing\n", want: "example.com/app"}, + {name: "missing", mod: "go 1.24\n", want: ""}, + } { + t.Run(tc.name, func(t *testing.T) { + assert.Equal(t, tc.want, modulePath([]byte(tc.mod))) + }) + } +} + +func TestFindModule(t *testing.T) { + dir := writeModule( + t, + "example.com/app", + map[string]string{"cmd/bundle/main.go": "package main\n"}, + ) + + mod, err := findModule(filepath.Join(dir, "cmd", "bundle")) + require.NoError(t, err) + assert.Equal(t, goModule{dir: dir, path: "example.com/app"}, mod) + + bare := t.TempDir() + mod, err = findModule(bare) + require.NoError(t, err) + assert.Equal(t, goModule{dir: bare}, mod, "without a go.mod the directory is the module root") +} + +func TestModuleRelPath(t *testing.T) { + dir := writeModule(t, "example.com/app", map[string]string{ + "main.go": "package main\n", + "dags/etl.go": "package dags\n", + "sub/nested.go": "package sub\n", + }) + mod := goModule{dir: dir, path: "example.com/app"} + + for _, tc := range []struct { + name string + reported string + want string + ok bool + }{ + {name: "absolute build path", reported: filepath.ToSlash(filepath.Join(dir, "dags", "etl.go")), want: "dags/etl.go", ok: true}, + {name: "trimpath", reported: "example.com/app/dags/etl.go", want: "dags/etl.go", ok: true}, + {name: "trimpath at root", reported: "example.com/app/main.go", want: "main.go", ok: true}, + {name: "absolute path outside the module", reported: "/home/u/go/pkg/mod/example.com/lib@v1.0.0/x.go"}, + {name: "another module", reported: "example.com/lib@v1.0.0/x.go"}, + {name: "module path prefix of another module", reported: "example.com/app/other@v1.0.0/x.go"}, + {name: "missing file", reported: "example.com/app/gone.go"}, + {name: "escapes the module", reported: "example.com/app/../x.go"}, + {name: "directory", reported: "example.com/app/dags"}, + {name: "empty", reported: ""}, + } { + t.Run(tc.name, func(t *testing.T) { + got, ok := moduleRelPath(tc.reported, mod) + assert.Equal(t, tc.ok, ok) + assert.Equal(t, tc.want, got) + }) + } +} + +func TestLayoutSources(t *testing.T) { + dir := writeModule(t, "example.com/app", map[string]string{ + "cmd/bundle/main.go": "package main\n", + "dags/b.go": "package dags // b\n", + "dags/a.go": "package dags // a\n", + }) + mod := goModule{dir: dir, path: "example.com/app"} + meta := airflowmetadata.Manifest{DagSourceFiles: map[string]string{ + "on_b": "example.com/app/dags/b.go", + "also_b": filepath.ToSlash(filepath.Join(dir, "dags", "b.go")), + "on_a": "example.com/app/dags/a.go", + "on_main": "example.com/app/cmd/bundle/main.go", + "foreign": "example.com/lib@v1.0.0/x.go", + }} + + var stderr bytes.Buffer + layout, region, err := layoutSources( + &stderr, + meta, + filepath.Join(dir, "cmd", "bundle", "main.go"), + mod, + ) + require.NoError(t, err) + + assert.Equal(t, "cmd/bundle/main.go", layout.entrypoint) + assert.Equal(t, map[string]string{ + "on_b": "dags/b.go", + "also_b": "dags/b.go", + "on_a": "dags/a.go", + "on_main": "cmd/bundle/main.go", + "foreign": "cmd/bundle/main.go", + }, layout.dagPaths) + + var paths []string + offset := 0 + for _, f := range layout.files { + paths = append(paths, f.path) + data, err := os.ReadFile(f.disk) + require.NoError(t, err) + assert.Equal(t, offset, f.offset, f.path) + assert.Equal(t, len(data), f.length, f.path) + assert.Equal(t, sha256Hex(data), f.sha256, f.path) + assert.Equal(t, data, region[f.offset:f.offset+f.length], f.path) + offset += f.length + } + assert.Equal(t, []string{"cmd/bundle/main.go", "dags/a.go", "dags/b.go"}, paths, + "entrypoint first, then the rest sorted, each file once") + assert.Len(t, region, offset) + + assert.Contains( + t, + stderr.String(), + `warning: source file "example.com/lib@v1.0.0/x.go" of dag "foreign"`, + ) + assert.Equal(t, 1, bytes.Count(stderr.Bytes(), []byte("warning:"))) +} + +func TestLayoutSources_EntrypointOutsideModule(t *testing.T) { + mod := goModule{dir: t.TempDir()} + elsewhere := filepath.Join(t.TempDir(), "main.go") + require.NoError(t, os.WriteFile(elsewhere, []byte("package main\n"), 0o644)) + + layout, _, err := layoutSources(&bytes.Buffer{}, airflowmetadata.Manifest{}, elsewhere, mod) + require.NoError(t, err) + assert.Equal(t, "main.go", layout.entrypoint) + assert.Empty(t, layout.dagPaths) +} + +func TestRunPack_EmbedsEntrypointWhenBundleHasNoDagSources(t *testing.T) { + dir := writeModule( + t, + "example.com/app", + map[string]string{"main.go": "package main\nfunc main() {}\n"}, + ) + exe := filepath.Join(dir, "prebuilt") + require.NoError(t, os.WriteFile(exe, []byte("prebuilt-binary-bytes"), 0o755)) + meta := filepath.Join(dir, "airflow-metadata.json") + require.NoError(t, os.WriteFile(meta, []byte( + `{"airflow_bundle_metadata_version":"1.0",`+ + `"sdk":{"language":"go","version":"0.1.0","supervisor_schema_version":"2026-06-16"},`+ + `"dags":{"my_dag":{"tasks":["t1"]}}}`, + ), 0o644)) + out := filepath.Join(t.TempDir(), "bundle") + + require.NoError(t, runPack(&bytes.Buffer{}, &bytes.Buffer{}, &packOptions{ + executable: exe, + source: filepath.Join(dir, "main.go"), + airflowMetadata: meta, + output: out, + })) + + m, region, _ := readPacked(t, out) + assert.Equal(t, "main.go", m.EntrypointPath) + assert.Empty(t, m.DagSourcePaths) + require.Len(t, m.Sources, 1) + assert.Equal(t, "package main\nfunc main() {}\n", string(region)) +} + +func TestRunPack_DagFileOutsideModuleFallsBackToEntrypoint(t *testing.T) { + dir := writeModule( + t, + "example.com/app", + map[string]string{"main.go": "package main\nfunc main() {}\n"}, + ) + exe := filepath.Join(dir, "prebuilt") + require.NoError(t, os.WriteFile(exe, []byte("prebuilt-binary-bytes"), 0o755)) + meta := filepath.Join(dir, "airflow-metadata.json") + require.NoError(t, os.WriteFile(meta, []byte( + `{"airflow_bundle_metadata_version":"1.0",`+ + `"sdk":{"language":"go","version":"0.1.0","supervisor_schema_version":"2026-06-16"},`+ + `"dags":{"py_dag":{"tasks":["t1"]}},`+ + `"dag_source_files":{"vendored":"/root/go/pkg/mod/example.com/lib@v1.0.0/dag.go"}}`, + ), 0o644)) + out := filepath.Join(t.TempDir(), "bundle") + + var stderr bytes.Buffer + require.NoError(t, runPack(&bytes.Buffer{}, &stderr, &packOptions{ + executable: exe, + source: filepath.Join(dir, "main.go"), + airflowMetadata: meta, + output: out, + })) + + m, _, _ := readPacked(t, out) + assert.Equal(t, map[string]string{"vendored": "main.go"}, m.DagSourcePaths) + require.Len(t, m.Sources, 1) + assert.Contains( + t, + stderr.String(), + `warning: source file "/root/go/pkg/mod/example.com/lib@v1.0.0/dag.go" of dag "vendored"`, + ) +} + +func TestRejectOutputAlias_EmbeddedSourceFile(t *testing.T) { + dir := t.TempDir() + exe := filepath.Join(dir, "exe") + entry := filepath.Join(dir, "main.go") + other := filepath.Join(dir, "dag.go") + for _, p := range []string{exe, entry, other} { + require.NoError(t, os.WriteFile(p, []byte("x"), 0o644)) + } + + err := rejectOutputAlias(other, exe, []string{entry, other}, "") + require.Error(t, err) + assert.Contains(t, err.Error(), "same file as the source") + + assert.NoError(t, rejectOutputAlias(filepath.Join(dir, "out"), exe, []string{entry, other}, "")) +} + +// Packs the multidag fixture through the real build path: three Dag files across two imported +// packages plus the entrypoint, with and without -trimpath. +func TestPack_MultiDagBundle(t *testing.T) { + if testing.Short() { + t.Skip("shells out to `go build`") + } + if _, err := exec.LookPath("go"); err != nil { + t.Skip("go toolchain not on PATH") + } + + for _, tc := range []struct { + name string + build []string + }{ + {name: "build paths"}, + {name: "trimpath", build: []string{"--", "-trimpath"}}, + } { + t.Run(tc.name, func(t *testing.T) { + const fixture = "cmd/airflow-go-pack/testdata/multidag" + out := filepath.Join(t.TempDir(), "bundle") + var stderr bytes.Buffer + cmd := newRootCmd() + cmd.SetArgs(append([]string{"./testdata/multidag", "--output", out}, tc.build...)) + cmd.SetOut(&bytes.Buffer{}) + cmd.SetErr(&stderr) + require.NoError(t, cmd.Execute()) + assert.NotContains(t, stderr.String(), "warning") + + m, region, _ := readPacked(t, out) + assert.Equal(t, fixture+"/main.go", m.EntrypointPath) + assert.Equal(t, map[string]string{ + "orders": fixture + "/main.go", + "reports": fixture + "/reports/reports.go", + "billing": fixture + "/factory/factory.go", + "shipping": fixture + "/factory/factory.go", + }, m.DagSourcePaths) + + var paths []string + for _, s := range m.Sources { + paths = append(paths, s.Path) + data, err := os.ReadFile(filepath.Join("..", "..", filepath.FromSlash(s.Path))) + require.NoError(t, err) + assert.Equal(t, data, region[s.Offset:s.Offset+s.Length], s.Path) + assert.Equal(t, sha256Hex(data), s.SHA256, s.Path) + } + assert.Equal(t, []string{ + fixture + "/main.go", + fixture + "/factory/factory.go", + fixture + "/reports/reports.go", + }, paths) + }) + } +} + +// A bundle with task handlers only has no Dag source of its own: just the entrypoint. +func TestPack_TaskHandlersOnlyBundle(t *testing.T) { + if testing.Short() { + t.Skip("shells out to `go build`") + } + if _, err := exec.LookPath("go"); err != nil { + t.Skip("go toolchain not on PATH") + } + + out := filepath.Join(t.TempDir(), "bundle") + cmd := newRootCmd() + cmd.SetArgs([]string{"./testdata/handlersonly", "--output", out}) + cmd.SetOut(&bytes.Buffer{}) + cmd.SetErr(&bytes.Buffer{}) + require.NoError(t, cmd.Execute()) + + m, _, metadata := readPacked(t, out) + assert.Equal(t, "cmd/airflow-go-pack/testdata/handlersonly/main.go", m.EntrypointPath) + assert.Empty(t, m.DagSourcePaths) + assert.Len(t, m.Sources, 1) + assert.Contains(t, string(metadata), "dag_source_paths: {}") +} + +// Identical inputs give a byte-identical bundle. +func TestPack_IsDeterministic(t *testing.T) { + if testing.Short() { + t.Skip("shells out to `go build`") + } + if _, err := exec.LookPath("go"); err != nil { + t.Skip("go toolchain not on PATH") + } + + exe := filepath.Join(t.TempDir(), "multidag") + goBuild(t, "./testdata/multidag", exe, runtime.GOOS, runtime.GOARCH) + + pack := func() string { + out := filepath.Join(t.TempDir(), "bundle") + cmd := newRootCmd() + cmd.SetArgs([]string{ + "--executable", exe, + "--source", "testdata/multidag/main.go", + "--output", out, + }) + cmd.SetOut(&bytes.Buffer{}) + cmd.SetErr(&bytes.Buffer{}) + require.NoError(t, cmd.Execute()) + return out + } + first, second := pack(), pack() + a, err := os.ReadFile(first) + require.NoError(t, err) + b, err := os.ReadFile(second) + require.NoError(t, err) + assert.Equal(t, a, b) +} diff --git a/go-sdk/cmd/airflow-go-pack/testdata/handlersonly/main.go b/go-sdk/cmd/airflow-go-pack/testdata/handlersonly/main.go new file mode 100644 index 0000000000000..e16dab50f8895 --- /dev/null +++ b/go-sdk/cmd/airflow-go-pack/testdata/handlersonly/main.go @@ -0,0 +1,36 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +// Command handlersonly is a bundle fixture for the packer tests. It registers task handlers and +// no Dag from airflow.Dag. +package main + +import ( + "log" + + "github.com/apache/airflow/go-sdk/airflow" +) + +func extract(airflow.Context) error { return nil } + +func main() { + bundle := airflow.Bundle() + bundle.Register(airflow.TaskHandler("py_etl", "extract", extract)) + if err := bundle.Serve(); err != nil { + log.Fatal(err) + } +} diff --git a/go-sdk/cmd/airflow-go-pack/testdata/multidag/factory/factory.go b/go-sdk/cmd/airflow-go-pack/testdata/multidag/factory/factory.go new file mode 100644 index 0000000000000..833e8e510aa65 --- /dev/null +++ b/go-sdk/cmd/airflow-go-pack/testdata/multidag/factory/factory.go @@ -0,0 +1,29 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +// Package factory builds Dags for several dag_ids from one file. +package factory + +import "github.com/apache/airflow/go-sdk/airflow" + +func run(airflow.Context) error { return nil } + +func New(dagID string) *airflow.DagRef { + dag := airflow.Dag(dagID) + dag.Task(run) + return dag +} diff --git a/go-sdk/cmd/airflow-go-pack/testdata/multidag/main.go b/go-sdk/cmd/airflow-go-pack/testdata/multidag/main.go new file mode 100644 index 0000000000000..64db46f26264d --- /dev/null +++ b/go-sdk/cmd/airflow-go-pack/testdata/multidag/main.go @@ -0,0 +1,49 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +// Command multidag is a bundle fixture for the packer tests. It declares Dags in its own file, +// in an imported package, and through a factory in a third package. +package main + +import ( + "log" + + "github.com/apache/airflow/go-sdk/airflow" + "github.com/apache/airflow/go-sdk/cmd/airflow-go-pack/testdata/multidag/factory" + "github.com/apache/airflow/go-sdk/cmd/airflow-go-pack/testdata/multidag/reports" +) + +func extract(airflow.Context) error { return nil } + +func main() { + bundle := airflow.Bundle() + + orders := airflow.Dag("orders") + orders.Task(extract) + + bundle.Register( + airflow.TaskHandler("py_etl", "extract", extract), + orders, + reports.Dag(), + factory.New("billing"), + factory.New("shipping"), + ) + + if err := bundle.Serve(); err != nil { + log.Fatal(err) + } +} diff --git a/go-sdk/cmd/airflow-go-pack/testdata/multidag/reports/reports.go b/go-sdk/cmd/airflow-go-pack/testdata/multidag/reports/reports.go new file mode 100644 index 0000000000000..7b432b41c731d --- /dev/null +++ b/go-sdk/cmd/airflow-go-pack/testdata/multidag/reports/reports.go @@ -0,0 +1,29 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +// Package reports declares a Dag in a file other than the bundle's entrypoint. +package reports + +import "github.com/apache/airflow/go-sdk/airflow" + +func build(airflow.Context) error { return nil } + +func Dag() *airflow.DagRef { + dag := airflow.Dag("reports") + dag.Task(build) + return dag +} From be71cdaec2e9e579ddb419aec596b1d22bdf22a5 Mon Sep 17 00:00:00 2001 From: ZHE YOU LIU Date: Tue, 6 Oct 2026 12:36:09 +0000 Subject: [PATCH 03/11] Print every embedded source file from airflow-go-pack inspect inspect --source cuts the source region by the manifest's sources index and prints each file under its own header. An index that does not fit the region is an error. --- go-sdk/cmd/airflow-go-pack/inspect.go | 58 ++++++++++++-- go-sdk/cmd/airflow-go-pack/inspect_test.go | 89 +++++++++++++++++++++- 2 files changed, 137 insertions(+), 10 deletions(-) diff --git a/go-sdk/cmd/airflow-go-pack/inspect.go b/go-sdk/cmd/airflow-go-pack/inspect.go index f9cca6d488b22..36c22c3c5850d 100644 --- a/go-sdk/cmd/airflow-go-pack/inspect.go +++ b/go-sdk/cmd/airflow-go-pack/inspect.go @@ -21,6 +21,7 @@ import ( "fmt" "github.com/spf13/cobra" + "gopkg.in/yaml.v3" "github.com/apache/airflow/go-sdk/internal/bundlefooter" ) @@ -29,7 +30,7 @@ func newInspectCmd() *cobra.Command { var showSource bool cmd := &cobra.Command{ Use: "inspect ", - Short: "Print the manifest (and optionally source) embedded in a bundle", + Short: "Print the manifest (and optionally source files) embedded in a bundle", Args: cobra.ExactArgs(1), RunE: func(cmd *cobra.Command, args []string) error { source, manifest, err := bundlefooter.Read(args[0]) @@ -38,10 +39,16 @@ func newInspectCmd() *cobra.Command { } out := cmd.OutOrStdout() if showSource { - fmt.Fprintln(out, "# --- source ---") - out.Write(source) - if len(source) > 0 && source[len(source)-1] != '\n' { - fmt.Fprintln(out) + files, err := embeddedSources(source, manifest) + if err != nil { + return err + } + for _, f := range files { + fmt.Fprintf(out, "# --- source: %s ---\n", f.path) + out.Write(f.data) + if len(f.data) > 0 && f.data[len(f.data)-1] != '\n' { + fmt.Fprintln(out) + } } fmt.Fprintln(out, "# --- manifest ---") } @@ -52,6 +59,45 @@ func newInspectCmd() *cobra.Command { return nil }, } - cmd.Flags().BoolVar(&showSource, "source", false, "also print the embedded source file") + cmd.Flags().BoolVar(&showSource, "source", false, "also print each embedded source file") return cmd } + +type embeddedSource struct { + path string + data []byte +} + +// embeddedSources cuts the source region into the files the manifest's sources index lists. +func embeddedSources(region, manifest []byte) ([]embeddedSource, error) { + var index struct { + Sources []struct { + Path string `yaml:"path"` + Offset int `yaml:"offset"` + Length int `yaml:"length"` + } `yaml:"sources"` + } + if err := yaml.Unmarshal(manifest, &index); err != nil { + return nil, fmt.Errorf("decoding manifest: %w", err) + } + if len(index.Sources) == 0 && len(region) > 0 { + return nil, fmt.Errorf( + "the source region has %d bytes but the manifest lists no sources", + len(region), + ) + } + files := make([]embeddedSource, 0, len(index.Sources)) + for _, src := range index.Sources { + if src.Offset < 0 || src.Length < 0 || src.Offset > len(region)-src.Length { + return nil, fmt.Errorf( + "manifest source %q (offset %d, length %d) does not fit the %d-byte source region", + src.Path, src.Offset, src.Length, len(region), + ) + } + files = append( + files, + embeddedSource{path: src.Path, data: region[src.Offset : src.Offset+src.Length]}, + ) + } + return files, nil +} diff --git a/go-sdk/cmd/airflow-go-pack/inspect_test.go b/go-sdk/cmd/airflow-go-pack/inspect_test.go index 8208136e80176..6692aeaab5c24 100644 --- a/go-sdk/cmd/airflow-go-pack/inspect_test.go +++ b/go-sdk/cmd/airflow-go-pack/inspect_test.go @@ -20,7 +20,9 @@ package main import ( "bytes" "os" + "os/exec" "path/filepath" + "strconv" "testing" "github.com/stretchr/testify/assert" @@ -28,21 +30,30 @@ import ( ) // inspect reads a bundle through bundlefooter.Read and prints the embedded -// manifest, prefixing the source too under --source. +// manifest, prefixing each embedded source file too under --source. func TestInspectCmd(t *testing.T) { dir := t.TempDir() exe := filepath.Join(dir, "input-bin") require.NoError(t, os.WriteFile(exe, []byte("binary-bytes"), 0o755)) - source := []byte("package main\n\nfunc main() {}\n") + entry := "package main\n\nfunc main() {}\n" + dag := "package dags" manifest := []byte( "airflow_bundle_metadata_version: \"1.0\"\n" + + "entrypoint_path: \"cmd/main.go\"\n" + + "sources:\n" + + " - path: \"cmd/main.go\"\n" + + " offset: 0\n" + + " length: " + strconv.Itoa(len(entry)) + "\n" + + " - path: \"dags/etl.go\"\n" + + " offset: " + strconv.Itoa(len(entry)) + "\n" + + " length: " + strconv.Itoa(len(dag)) + "\n" + "dags:\n" + " my_dag:\n" + " tasks:\n" + " - \"t1\"\n", ) bundle := filepath.Join(dir, "bundle") - require.NoError(t, writeBundle(exe, bundle, source, manifest)) + require.NoError(t, writeBundle(exe, bundle, []byte(entry+dag), manifest)) for _, tc := range []struct { name string @@ -57,7 +68,8 @@ func TestInspectCmd(t *testing.T) { { name: "with source", args: []string{"--source", bundle}, - expect: "# --- source ---\n" + string(source) + + expect: "# --- source: cmd/main.go ---\n" + entry + + "# --- source: dags/etl.go ---\n" + dag + "\n" + "# --- manifest ---\n" + string(manifest), }, } { @@ -72,3 +84,72 @@ func TestInspectCmd(t *testing.T) { }) } } + +func TestEmbeddedSources_RejectsIndexOutsideRegion(t *testing.T) { + for _, tc := range []struct { + name string + region string + manifest string + wantErr string + }{ + { + name: "past the end", + region: "abc", + manifest: "sources:\n - {path: a.go, offset: 2, length: 5}\n", + wantErr: `source "a.go" (offset 2, length 5) does not fit the 3-byte source region`, + }, + { + name: "negative offset", + region: "abc", + manifest: "sources:\n - {path: a.go, offset: -1, length: 2}\n", + wantErr: `source "a.go"`, + }, + { + name: "region without an index", + region: "abc", + manifest: "dags: {}\n", + wantErr: "lists no sources", + }, + } { + t.Run(tc.name, func(t *testing.T) { + _, err := embeddedSources([]byte(tc.region), []byte(tc.manifest)) + require.Error(t, err) + assert.Contains(t, err.Error(), tc.wantErr) + }) + } + + files, err := embeddedSources(nil, []byte("dags: {}\n")) + require.NoError(t, err) + assert.Empty(t, files) +} + +// Inspect prints every file that packing the multidag fixture embedded. +func TestInspectCmd_PrintsEachPackedFile(t *testing.T) { + if testing.Short() { + t.Skip("shells out to `go build`") + } + if _, err := exec.LookPath("go"); err != nil { + t.Skip("go toolchain not on PATH") + } + + out := filepath.Join(t.TempDir(), "bundle") + pack := newRootCmd() + pack.SetArgs([]string{"./testdata/multidag", "--output", out}) + pack.SetOut(&bytes.Buffer{}) + pack.SetErr(&bytes.Buffer{}) + require.NoError(t, pack.Execute()) + + inspect := newInspectCmd() + var printed bytes.Buffer + inspect.SetOut(&printed) + inspect.SetArgs([]string{"--source", out}) + require.NoError(t, inspect.Execute()) + for _, name := range []string{"main.go", "factory/factory.go", "reports/reports.go"} { + assert.Contains( + t, + printed.String(), + "# --- source: cmd/airflow-go-pack/testdata/multidag/"+name+" ---\n", + ) + } + assert.Contains(t, printed.String(), "# --- manifest ---\n") +} From 5b432ffb08d5f17ca066ce03187fb54f55b84d4f Mon Sep 17 00:00:00 2001 From: ZHE YOU LIU Date: Tue, 6 Oct 2026 12:36:32 +0000 Subject: [PATCH 04/11] Document per-Dag source files in Go bundles The Go SDK README, the Go language SDK page and the example Justfile now say a bundle embeds the entrypoint plus the source file of each native Dag. --- .../docs/authoring-and-scheduling/language-sdks/go.rst | 5 +++-- go-sdk/README.md | 4 +++- go-sdk/example/bundle/Justfile | 4 ++-- 3 files changed, 8 insertions(+), 5 deletions(-) diff --git a/airflow-core/docs/authoring-and-scheduling/language-sdks/go.rst b/airflow-core/docs/authoring-and-scheduling/language-sdks/go.rst index 7648c141531b3..f2d7613e356e9 100644 --- a/airflow-core/docs/authoring-and-scheduling/language-sdks/go.rst +++ b/airflow-core/docs/authoring-and-scheduling/language-sdks/go.rst @@ -28,7 +28,8 @@ to a compiled Go *bundle* that is launched by :class:`~airflow.sdk.coordinators.executable.ExecutableCoordinator` for each task instance. Because Go is a compiled language, every task must be compiled ahead of time and registered inside a single, -self-contained native executable called a **bundle**. The bundle also embeds its Dag source and a metadata +self-contained native executable called a **bundle**. The bundle also embeds the source file of each Dag built with ``airflow.Dag`` +plus the entrypoint file (the one with ``func main``), and a metadata manifest (the ``dag_id`` and ``task_id`` map) in a footer appended to the executable, so the executable *is* the bundle: one runnable file to ship, with no separate manifest or archive. The :ref:`airflow-go-pack ` tool builds and packs that bundle. @@ -529,7 +530,7 @@ surface as Go values when read back via ``GetXCom``. Building and packaging ---------------------- -A plain ``go build`` produces a runnable binary, but a *deployable* bundle (binary + embedded source + +A plain ``go build`` produces a runnable binary, but a *deployable* bundle (binary + embedded source files + manifest) must be produced with ``airflow-go-pack``. The packer compiles the bundle and appends the embedded metadata footer, so the coordinator can read its ``dag_id``\ s without executing the binary, producing a single runnable file. The on-disk format the packer emits (the ``AFBNDL01`` footer and the diff --git a/go-sdk/README.md b/go-sdk/README.md index f099bee045efe..a865e52f24bef 100644 --- a/go-sdk/README.md +++ b/go-sdk/README.md @@ -38,7 +38,9 @@ Python tasks are imported and run in-process. Go is compiled, so the model is di A single binary that bundles one or more Dags' task functions is called a **bundle**. You build one with the SDK's packer, `airflow-go-pack`, which compiles your code and appends a metadata footer (the manifest -of `dag_id`s and `task_id`s, plus the Dag source) to the executable. The result is a **self-contained +of `dag_id`s and `task_id`s, plus the source files) to the executable. The source files are the +entrypoint (the file with `func main`) and the file that declares each Dag built with `airflow.Dag`. +A Dag declared outside your module, such as in a dependency, uses the entrypoint as its source. The result is a **self-contained executable bundle**: a single runnable file that *is* the bundle, with no separate manifest or archive to ship alongside it. diff --git a/go-sdk/example/bundle/Justfile b/go-sdk/example/bundle/Justfile index 07df82a2280a6..97ae2025c14c4 100644 --- a/go-sdk/example/bundle/Justfile +++ b/go-sdk/example/bundle/Justfile @@ -26,8 +26,8 @@ build: pack # One-step build + pack. The single `go tool airflow-go-pack` # invocation runs `go build` internally, queries the binary for its -# DAG/task identity via --airflow-metadata, and appends the source plus -# airflow-metadata.yaml plus AFBNDL01 trailer. The output is a single +# DAG/task identity via --airflow-metadata, and appends the source files +# plus airflow-metadata.yaml plus AFBNDL01 trailer. The output is a single # self-contained executable bundle, named after the bundle's package # directory and written to the current directory. Drop it into the # Dag bundle named by ExecutableCoordinator's task_handler_bundle_name From bce7e6b545b11e4d54f741b01516f048cbbad045 Mon Sep 17 00:00:00 2001 From: ZHE YOU LIU Date: Wed, 7 Oct 2026 09:17:54 +0000 Subject: [PATCH 05/11] Go SDK: run the tasks of Dags built with airflow.Dag Serve looked tasks up among the task handlers only, so every task of a native Go Dag would be reported as not registered once Airflow runs native Go Dags. --- go-sdk/airflow/bundle.go | 28 ++++++++++++++++++++++++++++ go-sdk/airflow/bundle_test.go | 27 +++++++++++++++++++++++++++ go-sdk/airflow/dag.go | 4 ++-- go-sdk/internal/bundle/doc.go | 9 +++++---- 4 files changed, 62 insertions(+), 6 deletions(-) diff --git a/go-sdk/airflow/bundle.go b/go-sdk/airflow/bundle.go index ee0c738038804..61bcd77aa8fd5 100644 --- a/go-sdk/airflow/bundle.go +++ b/go-sdk/airflow/bundle.go @@ -229,6 +229,25 @@ func (m *dagMap) ListDagSourceFiles() map[string]string { return files } +// lookupTask returns the Go function of a task of a registered Dag. A task from TriggerDagRun has +// none, because a Python worker runs it. +func (m *dagMap) lookupTask(dagID, taskID string) (bundle.Task, bool) { + m.mu.Lock() + dag, ok := m.dags[dagID] + m.mu.Unlock() + if !ok { + return nil, false + } + + dag.mu.Lock() + defer dag.mu.Unlock() + ref, ok := dag.tasksByID[taskID] + if !ok || ref.task == nil { + return nil, false + } + return ref.task, true +} + func (m *dagMap) has(dagID string) bool { m.mu.Lock() defer m.mu.Unlock() @@ -286,3 +305,12 @@ func (s coordinatorSource) SerializeDags(fileloc, relativeFileloc string) []bund func (s coordinatorSource) ListDagSourceFiles() map[string]string { return s.dags.ListDagSourceFiles() } + +// LookupTask finds a task of a Dag from airflow.Dag, then a task handler. A dag_id belongs to only +// one of the two, as Register checks. +func (s coordinatorSource) LookupTask(dagID, taskID string) (bundle.Task, bool) { + if task, ok := s.dags.lookupTask(dagID, taskID); ok { + return task, true + } + return s.taskHandlerMap.LookupTask(dagID, taskID) +} diff --git a/go-sdk/airflow/bundle_test.go b/go-sdk/airflow/bundle_test.go index d6c2da67f5120..1ddc3b2b1097f 100644 --- a/go-sdk/airflow/bundle_test.go +++ b/go-sdk/airflow/bundle_test.go @@ -453,3 +453,30 @@ func TestSerializeDagsLeavesOutADagThatRegisterRejected(t *testing.T) { assert.Empty(t, b.dags.serialize("/bundles/go/etl", "etl")) } + +func TestServeLooksUpTheTasksOfDagsAndTaskHandlers(t *testing.T) { + dag := Dag("native_etl") + dag.Task(ping, TaskSpec{TaskID: "extract"}) + dag.Task( + TriggerDagRun(TriggerDagRunSpec{DagID: "downstream_etl"}), + TaskSpec{TaskID: "trigger"}, + ) + b := Bundle() + b.Register(dag, TaskHandler("py_etl", "load", noop)) + source := coordinatorSource{&b.taskHandlers, &b.dags} + + for _, id := range [][2]string{{"native_etl", "extract"}, {"py_etl", "load"}} { + task, ok := source.LookupTask(id[0], id[1]) + assert.True(t, ok, "%s.%s", id[0], id[1]) + assert.NotNil(t, task) + } + for _, id := range [][2]string{ + {"native_etl", "trigger"}, + {"native_etl", "load"}, + {"py_etl", "extract"}, + {"unknown", "extract"}, + } { + _, ok := source.LookupTask(id[0], id[1]) + assert.False(t, ok, "%s.%s", id[0], id[1]) + } +} diff --git a/go-sdk/airflow/dag.go b/go-sdk/airflow/dag.go index 0e4736ee5fd63..31950d51a0c51 100644 --- a/go-sdk/airflow/dag.go +++ b/go-sdk/airflow/dag.go @@ -77,8 +77,8 @@ type DagRef struct { // The bundle embeds the source file that calls Dag, so call it from the file that declares the Dag. // A Dag built in a factory function belongs to the file of the factory. // -// [BundleRef.Serve] sends the registered Dags to the Dag processor, but does not yet list them in -// the --airflow-metadata manifest or run their tasks. +// [BundleRef.Serve] sends the registered Dags to the Dag processor and runs their tasks, but does +// not list them in the dags of the --airflow-metadata manifest. // // Dag panics if it gets more than one DagSpec, or if the DagSpec has a value that Python rejects // when it builds or validates a Dag: diff --git a/go-sdk/internal/bundle/doc.go b/go-sdk/internal/bundle/doc.go index 8a77698fe70a8..0222ffffe0cc3 100644 --- a/go-sdk/internal/bundle/doc.go +++ b/go-sdk/internal/bundle/doc.go @@ -16,9 +16,10 @@ // under the License. // Package bundle defines what the coordinator runtime needs from a bundle: the -// tasks it looks up and runs, the Dag and task ids it lists in the manifest, and -// the serialized Dags it sends to the Dag processor. +// tasks it looks up and runs, the Dag and task ids and the Dag source files it +// lists in the manifest, and the serialized Dags it sends to the Dag processor. // -// Package airflow builds the tasks and ids from the task handlers a bundle -// registers, and the serialized Dags from its Dags. +// Package airflow builds the tasks from both the task handlers and the Dags a +// bundle registers, the ids from its task handlers, and the source files and +// serialized Dags from its Dags. package bundle From 3750a31e3dd2965a83463043ca1e3ca3af0ff458 Mon Sep 17 00:00:00 2001 From: ZHE YOU LIU Date: Wed, 7 Oct 2026 15:59:16 +0000 Subject: [PATCH 06/11] Go SDK: pack an entrypoint at the module root The module root itself now counts as inside the module, so a main.go at the top of a plain go mod init project packs. --- go-sdk/cmd/airflow-go-pack/sources.go | 3 ++- go-sdk/cmd/airflow-go-pack/sources_test.go | 14 ++++++++++++++ 2 files changed, 16 insertions(+), 1 deletion(-) diff --git a/go-sdk/cmd/airflow-go-pack/sources.go b/go-sdk/cmd/airflow-go-pack/sources.go index a49746a2b6f8a..3455b73cb17ea 100644 --- a/go-sdk/cmd/airflow-go-pack/sources.go +++ b/go-sdk/cmd/airflow-go-pack/sources.go @@ -86,7 +86,8 @@ func listModule(dir string) (goModule, error) { lines := splitNonEmpty(stdout.String()) var best goModule for i := 0; i+1 < len(lines); i += 2 { - if _, ok := relativeTo(lines[i], absDir); ok && len(lines[i]) > len(best.dir) { + _, inside := relativeTo(lines[i], absDir) + if (inside || lines[i] == absDir) && len(lines[i]) > len(best.dir) { best = goModule{dir: lines[i], path: lines[i+1]} } } diff --git a/go-sdk/cmd/airflow-go-pack/sources_test.go b/go-sdk/cmd/airflow-go-pack/sources_test.go index 96a87c423cb04..1addf0d262345 100644 --- a/go-sdk/cmd/airflow-go-pack/sources_test.go +++ b/go-sdk/cmd/airflow-go-pack/sources_test.go @@ -75,6 +75,20 @@ func writeModule(t *testing.T, modulePath string, files map[string]string) strin return dir } +func TestListModule_EntrypointAtModuleRoot(t *testing.T) { + if _, err := exec.LookPath("go"); err != nil { + t.Skip("go toolchain not on PATH") + } + t.Setenv("GOWORK", "off") + dir := writeModule(t, "example.com/app", map[string]string{ + "main.go": "package main\nfunc main() {}\n", + }) + + mod, err := listModule(dir) + require.NoError(t, err) + assert.Equal(t, "example.com/app", mod.path) +} + func TestModulePath(t *testing.T) { for _, tc := range []struct { name string From 4bc863d0f2cc37d71b228eedca27abe7b7be0cb5 Mon Sep 17 00:00:00 2001 From: ZHE YOU LIU Date: Wed, 7 Oct 2026 16:00:05 +0000 Subject: [PATCH 07/11] Go SDK: quote dag_id keys in the bundle manifest Dag IDs such as 2024, on or null were written unquoted and read back as int, bool or None by a YAML parser. --- go-sdk/cmd/airflow-go-pack/pack.go | 16 +++---- .../airflow-go-pack/pack_integration_test.go | 12 ++--- go-sdk/cmd/airflow-go-pack/pack_test.go | 48 ++++++++++++------- 3 files changed, 46 insertions(+), 30 deletions(-) diff --git a/go-sdk/cmd/airflow-go-pack/pack.go b/go-sdk/cmd/airflow-go-pack/pack.go index 8d96330b864d2..1f0f2db85ba1c 100644 --- a/go-sdk/cmd/airflow-go-pack/pack.go +++ b/go-sdk/cmd/airflow-go-pack/pack.go @@ -505,7 +505,7 @@ func renderManifest(meta airflowmetadata.Manifest, layout sourceLayout) ([]byte, taskItems = append(taskItems, quotedScalar(t)) } dagsNode.Content = append(dagsNode.Content, - scalar(id), + quotedScalar(id), &yaml.Node{ Kind: yaml.MappingNode, Content: []*yaml.Node{ @@ -525,7 +525,7 @@ func renderManifest(meta airflowmetadata.Manifest, layout sourceLayout) ([]byte, for _, id := range dagPaths { dagPathsNode.Content = append( dagPathsNode.Content, - scalar(id), + quotedScalar(id), quotedScalar(layout.dagPaths[id]), ) } @@ -578,8 +578,8 @@ func renderManifest(meta airflowmetadata.Manifest, layout sourceLayout) ([]byte, return buf.Bytes(), nil } -// scalar emits a plain (unquoted) node. It is used for structural keys -// (e.g. "sdk", "tasks") and for the Dag ID mapping keys. +// scalar emits a plain (unquoted) node. It is for structural keys only (e.g. "sdk", "tasks"). +// Dag IDs are data, so they go through quotedScalar. func scalar(value string) *yaml.Node { return &yaml.Node{Kind: yaml.ScalarNode, Value: value} } @@ -588,10 +588,10 @@ func intScalar(value int) *yaml.Node { return &yaml.Node{Kind: yaml.ScalarNode, Tag: "!!int", Value: strconv.Itoa(value)} } -// quotedScalar emits a double-quoted node. Data-bearing string *values* — task -// IDs, the source paths, and the SDK fields — go through this so a value -// that looks like a number, bool, or date (e.g. a task named "123" or "true") -// round-trips as a string rather than being retyped by the YAML parser. +// quotedScalar emits a double-quoted node. Data-bearing strings, such as Dag IDs, task IDs, +// the source paths, and the SDK fields, go through this so a value that looks like a number, +// bool, or date (e.g. a Dag named "2024" or "on") round-trips as a string rather than being +// retyped by the YAML parser. func quotedScalar(value string) *yaml.Node { return &yaml.Node{Kind: yaml.ScalarNode, Value: value, Style: yaml.DoubleQuotedStyle} } diff --git a/go-sdk/cmd/airflow-go-pack/pack_integration_test.go b/go-sdk/cmd/airflow-go-pack/pack_integration_test.go index f4ab7e4ac7b6d..019fc6c3fd23a 100644 --- a/go-sdk/cmd/airflow-go-pack/pack_integration_test.go +++ b/go-sdk/cmd/airflow-go-pack/pack_integration_test.go @@ -153,18 +153,18 @@ sources: length: ` + strconv.Itoa(len(srcBytes)) + ` sha256: "` + sha256Hex(srcBytes) + `" dags: - concurrent_xcom_dag: + "concurrent_xcom_dag": tasks: - "pull_xcoms_concurrently" - simple_dag: + "simple_dag": tasks: - "extract" - "transform" - "load" - task_state_dag: + "task_state_dag": tasks: - "roundtrip_task_state" - taskflow_binding_dag: + "taskflow_binding_dag": tasks: - "make_config" - "make_numbers" @@ -178,7 +178,7 @@ dags: - "via_flat_map" - "via_struct_map" - "via_plain_map" - variable_write_dag: + "variable_write_dag": tasks: - "write_and_delete_variable" ` @@ -232,7 +232,7 @@ func TestPack_CrossCompileBuildModeForwardsFlags(t *testing.T) { source, metadata, err := bundlefooter.Read(outPath) require.NoError(t, err) - assert.Contains(t, string(metadata), "simple_dag:", + assert.Contains(t, string(metadata), `"simple_dag":`, "manifest must be read from the host introspection build") // Independently build the target-arch artefact with the same forwarded diff --git a/go-sdk/cmd/airflow-go-pack/pack_test.go b/go-sdk/cmd/airflow-go-pack/pack_test.go index f90ab92b1480a..474bed85d5dd5 100644 --- a/go-sdk/cmd/airflow-go-pack/pack_test.go +++ b/go-sdk/cmd/airflow-go-pack/pack_test.go @@ -29,6 +29,7 @@ import ( "github.com/spf13/cobra" "github.com/stretchr/testify/assert" "github.com/stretchr/testify/require" + "gopkg.in/yaml.v3" "github.com/apache/airflow/go-sdk/internal/airflowmetadata" "github.com/apache/airflow/go-sdk/internal/bundlefooter" @@ -71,8 +72,8 @@ sdk: supervisor_schema_version: "2026-06-16" entrypoint_path: "cmd/bundle/main.go" dag_source_paths: - alpha_dag: "dags/alpha.go" - zeta_dag: "cmd/bundle/main.go" + "alpha_dag": "dags/alpha.go" + "zeta_dag": "cmd/bundle/main.go" sources: - path: "cmd/bundle/main.go" offset: 0 @@ -83,10 +84,10 @@ sources: length: 20 sha256: "bb" dags: - alpha_dag: + "alpha_dag": tasks: - "x" - zeta_dag: + "zeta_dag": tasks: - "a" - "b" @@ -94,9 +95,9 @@ dags: assert.Equal(t, expected, string(got1)) } -// Values (task IDs, source paths, SDK fields) are quoted so a scalar-looking value -// stays a string; Dag ID keys stay plain scalars. -func TestRenderManifest_QuotesValuesNotKeys(t *testing.T) { +// Values (task IDs, source paths, SDK fields) and Dag ID keys are quoted so a scalar-looking +// string stays a string when a YAML parser reads the manifest back. +func TestRenderManifest_QuotesDagIDs(t *testing.T) { meta := airflowmetadata.Manifest{ AirflowBundleMetadataVersion: "1.0", SDK: airflowmetadata.SDK{ @@ -105,19 +106,34 @@ func TestRenderManifest_QuotesValuesNotKeys(t *testing.T) { SupervisorSchemaVersion: "2026-06-16", }, Dags: map[string]airflowmetadata.Dag{ - "my_dag": {Tasks: []string{"123", "true"}}, + "2024": {Tasks: []string{"123", "true"}}, + "on": {Tasks: []string{"t1"}}, }, } + layout := sourceLayout{ + entrypoint: "main.go", + dagPaths: map[string]string{"2024": "main.go", "on": "main.go"}, + files: []sourceFile{{path: "main.go", offset: 0, length: 10, sha256: "aa"}}, + } - got, err := renderManifest(meta, testLayout) + got, err := renderManifest(meta, layout) require.NoError(t, err) - // Task values that look like scalars are quoted. assert.Contains(t, string(got), `- "123"`) assert.Contains(t, string(got), `- "true"`) - // The Dag ID key is a plain scalar, not quoted. - assert.Contains(t, string(got), "\n my_dag:\n") - assert.NotContains(t, string(got), `"my_dag"`) + assert.Contains(t, string(got), "\n \"2024\":\n") + assert.Contains(t, string(got), "\n \"on\":\n") + assert.Contains(t, string(got), `"2024": "main.go"`) + assert.Contains(t, string(got), `"on": "main.go"`) + + var decoded struct { + Dags map[any]any `yaml:"dags"` + DagSourcePaths map[any]any `yaml:"dag_source_paths"` + } + require.NoError(t, yaml.Unmarshal(got, &decoded)) + assert.Contains(t, decoded.Dags, "2024") + assert.Contains(t, decoded.Dags, "on") + assert.Equal(t, map[any]any{"2024": "main.go", "on": "main.go"}, decoded.DagSourcePaths) } func TestRenderManifest_EmptyDags(t *testing.T) { @@ -389,7 +405,7 @@ func TestRunPack_UsesMetadataFile(t *testing.T) { assert.Contains( t, string(gotMeta), - "my_dag:", + `"my_dag":`, "Dag from --airflow-metadata must appear in the manifest", ) @@ -423,7 +439,7 @@ func TestRunPack_PacksABundleWithOnlyNativeDags(t *testing.T) { _, gotMeta, err := bundlefooter.Read(out) require.NoError(t, err) - assert.Contains(t, string(gotMeta), `native_dag: "main.go"`) + assert.Contains(t, string(gotMeta), `"native_dag": "main.go"`) } // --airflow-metadata also accepts a YAML manifest, not only the JSON the @@ -461,7 +477,7 @@ func TestRunPack_AcceptsYAMLMetadataFile(t *testing.T) { assert.Contains( t, string(gotMeta), - "yaml_dag:", + `"yaml_dag":`, "Dag from a YAML --airflow-metadata file must appear in the manifest", ) } From f0004dc79c59109c5ae00c72a303e743dae1f392 Mon Sep 17 00:00:00 2001 From: ZHE YOU LIU Date: Wed, 7 Oct 2026 16:00:23 +0000 Subject: [PATCH 08/11] Go SDK: count native Dags in the pack summary The summary showed dags=0 for a bundle with only airflow.Dag Dags, so it now reports task handler and native Dags separately. --- go-sdk/cmd/airflow-go-pack/pack.go | 4 ++-- go-sdk/cmd/airflow-go-pack/pack_test.go | 5 ++++- 2 files changed, 6 insertions(+), 3 deletions(-) diff --git a/go-sdk/cmd/airflow-go-pack/pack.go b/go-sdk/cmd/airflow-go-pack/pack.go index 1f0f2db85ba1c..c09cbe87e1e6d 100644 --- a/go-sdk/cmd/airflow-go-pack/pack.go +++ b/go-sdk/cmd/airflow-go-pack/pack.go @@ -198,8 +198,8 @@ func runPack(stdout, stderr io.Writer, opts *packOptions) error { return err } - fmt.Fprintf(stdout, "Wrote bundle %s (sdk=%s/%s, dags=%d)\n", - output, meta.SDK.Language, meta.SDK.Version, len(meta.Dags)) + fmt.Fprintf(stdout, "Wrote bundle %s (sdk=%s/%s, task_handler_dags=%d, native_dags=%d)\n", + output, meta.SDK.Language, meta.SDK.Version, len(meta.Dags), len(meta.DagSourceFiles)) return nil } diff --git a/go-sdk/cmd/airflow-go-pack/pack_test.go b/go-sdk/cmd/airflow-go-pack/pack_test.go index 474bed85d5dd5..992e1e6da8709 100644 --- a/go-sdk/cmd/airflow-go-pack/pack_test.go +++ b/go-sdk/cmd/airflow-go-pack/pack_test.go @@ -18,6 +18,7 @@ package main import ( + "bytes" "errors" "io" "os" @@ -429,7 +430,8 @@ func TestRunPack_PacksABundleWithOnlyNativeDags(t *testing.T) { ), 0o644)) out := filepath.Join(dir, "bundle") - err := runPack(io.Discard, io.Discard, &packOptions{ + var stdout bytes.Buffer + err := runPack(&stdout, io.Discard, &packOptions{ executable: exe, source: source, airflowMetadata: meta, @@ -440,6 +442,7 @@ func TestRunPack_PacksABundleWithOnlyNativeDags(t *testing.T) { _, gotMeta, err := bundlefooter.Read(out) require.NoError(t, err) assert.Contains(t, string(gotMeta), `"native_dag": "main.go"`) + assert.Contains(t, stdout.String(), "task_handler_dags=0, native_dags=1") } // --airflow-metadata also accepts a YAML manifest, not only the JSON the From 0d5f0e31813fc42ad8e895e3de763f839758b734 Mon Sep 17 00:00:00 2001 From: ZHE YOU LIU Date: Wed, 7 Oct 2026 16:00:36 +0000 Subject: [PATCH 09/11] Go SDK: tell the user to repack a bundle without a sources index The error now says to repack the bundle with this airflow-go-pack instead of only describing the mismatch. --- go-sdk/cmd/airflow-go-pack/inspect.go | 3 ++- go-sdk/cmd/airflow-go-pack/inspect_test.go | 2 +- 2 files changed, 3 insertions(+), 2 deletions(-) diff --git a/go-sdk/cmd/airflow-go-pack/inspect.go b/go-sdk/cmd/airflow-go-pack/inspect.go index 36c22c3c5850d..9b0eb192ae72a 100644 --- a/go-sdk/cmd/airflow-go-pack/inspect.go +++ b/go-sdk/cmd/airflow-go-pack/inspect.go @@ -82,7 +82,8 @@ func embeddedSources(region, manifest []byte) ([]embeddedSource, error) { } if len(index.Sources) == 0 && len(region) > 0 { return nil, fmt.Errorf( - "the source region has %d bytes but the manifest lists no sources", + "the manifest lists no sources for the %d-byte source region; "+ + "repack the bundle with this airflow-go-pack", len(region), ) } diff --git a/go-sdk/cmd/airflow-go-pack/inspect_test.go b/go-sdk/cmd/airflow-go-pack/inspect_test.go index 6692aeaab5c24..696698950c6c4 100644 --- a/go-sdk/cmd/airflow-go-pack/inspect_test.go +++ b/go-sdk/cmd/airflow-go-pack/inspect_test.go @@ -108,7 +108,7 @@ func TestEmbeddedSources_RejectsIndexOutsideRegion(t *testing.T) { name: "region without an index", region: "abc", manifest: "dags: {}\n", - wantErr: "lists no sources", + wantErr: "repack the bundle", }, } { t.Run(tc.name, func(t *testing.T) { From 43d3315a03e9691084ee42f8a7ebf65a835002dc Mon Sep 17 00:00:00 2001 From: ZHE YOU LIU Date: Wed, 7 Oct 2026 16:00:55 +0000 Subject: [PATCH 10/11] Go SDK: test the per-Dag source alias check through runPack The old test called rejectOutputAlias directly, so removing the call in runPack for per-Dag files kept the suite green. --- go-sdk/cmd/airflow-go-pack/sources_test.go | 38 +++++++++++++++------- 1 file changed, 26 insertions(+), 12 deletions(-) diff --git a/go-sdk/cmd/airflow-go-pack/sources_test.go b/go-sdk/cmd/airflow-go-pack/sources_test.go index 1addf0d262345..3abe07c8b94fe 100644 --- a/go-sdk/cmd/airflow-go-pack/sources_test.go +++ b/go-sdk/cmd/airflow-go-pack/sources_test.go @@ -21,10 +21,12 @@ import ( "bytes" "crypto/sha256" "encoding/hex" + "io" "os" "os/exec" "path/filepath" "runtime" + "strconv" "testing" "github.com/stretchr/testify/assert" @@ -289,20 +291,32 @@ func TestRunPack_DagFileOutsideModuleFallsBackToEntrypoint(t *testing.T) { ) } -func TestRejectOutputAlias_EmbeddedSourceFile(t *testing.T) { - dir := t.TempDir() - exe := filepath.Join(dir, "exe") - entry := filepath.Join(dir, "main.go") - other := filepath.Join(dir, "dag.go") - for _, p := range []string{exe, entry, other} { - require.NoError(t, os.WriteFile(p, []byte("x"), 0o644)) - } +func TestRunPack_RejectsOutputAliasingDagSourceFile(t *testing.T) { + dir := writeModule(t, "example.com/app", map[string]string{ + "main.go": "package main\nfunc main() {}\n", + "dags/reports.go": "package dags\n", + }) + exe := filepath.Join(dir, "prebuilt") + require.NoError(t, os.WriteFile(exe, []byte("prebuilt-binary-bytes"), 0o755)) + reports := filepath.Join(dir, "dags", "reports.go") + meta := filepath.Join(dir, "airflow-metadata.json") + require.NoError(t, os.WriteFile(meta, []byte( + `{"airflow_bundle_metadata_version":"1.0",`+ + `"sdk":{"language":"go","version":"0.1.0","supervisor_schema_version":"2026-06-16"},`+ + `"dags":{},"dag_source_files":{"reports":`+strconv.Quote(reports)+`}}`, + ), 0o644)) - err := rejectOutputAlias(other, exe, []string{entry, other}, "") - require.Error(t, err) - assert.Contains(t, err.Error(), "same file as the source") + err := runPack(io.Discard, io.Discard, &packOptions{ + executable: exe, + source: filepath.Join(dir, "main.go"), + airflowMetadata: meta, + output: reports, + }) + require.ErrorContains(t, err, "same file as the source") - assert.NoError(t, rejectOutputAlias(filepath.Join(dir, "out"), exe, []string{entry, other}, "")) + got, readErr := os.ReadFile(reports) + require.NoError(t, readErr) + assert.Equal(t, "package dags\n", string(got)) } // Packs the multidag fixture through the real build path: three Dag files across two imported From 0e71e05dcf865740ae6a70e76cd6dbb497c6d4b8 Mon Sep 17 00:00:00 2001 From: ZHE YOU LIU Date: Wed, 7 Oct 2026 16:01:09 +0000 Subject: [PATCH 11/11] Document how Go maps a factory-built Dag to its source Go records the file that calls airflow.Dag, so a Dag built in a factory maps to the factory's file. --- .../0015-per-dag-source-in-bundle-artifact.md | 13 ++++++++----- 1 file changed, 8 insertions(+), 5 deletions(-) diff --git a/airflow-core/adr/lang-sdk/0015-per-dag-source-in-bundle-artifact.md b/airflow-core/adr/lang-sdk/0015-per-dag-source-in-bundle-artifact.md index 90c48b593f776..602066d958f8c 100644 --- a/airflow-core/adr/lang-sdk/0015-per-dag-source-in-bundle-artifact.md +++ b/airflow-core/adr/lang-sdk/0015-per-dag-source-in-bundle-artifact.md @@ -95,6 +95,9 @@ packers adopt the same shape — per-Dag source resolution, de-duplication by pa always-embedded entrypoint fallback — so a bundle reader treats every language the same way and the Code tab behaves identically regardless of which SDK produced the artifact. +Go has no module evaluation, so the Go SDK records the file that calls `airflow.Dag`. A Dag built in +a factory function maps to the factory's file, not to the file that calls the factory. + ## What a reader returns | the Task is… | reader returns | @@ -113,11 +116,11 @@ Code tab behaves identically regardless of which SDK produced the artifact. `DagCode` → the Code tab, deferred to [ADR-0010](0010-native-dag-processing.md)'s open question and future work) sees one contract. - Generated Dags (built in a loop or a factory during module evaluation) are resolved to their - generator file like any other Dag — they are fully supported, not a fallback case. Best-effort - resolution only bites a Dag built in a detached callback after its module finished, which then - shows the entrypoint — a real file in the bundle rather than a wrong one. This mirrors the source - view already being best effort for Python factory-function Dags - ([ADR-0006](0006-no-lang-sdk-source-display.md), "Why Not" #3). + generator file (for Go, the file that calls `airflow.Dag`) like any other Dag — they are fully + supported, not a fallback case. Best-effort resolution only bites a Dag built in a detached + callback after its module finished, which then shows the entrypoint — a real file in the bundle + rather than a wrong one. This mirrors the source view already being best effort for Python + factory-function Dags ([ADR-0006](0006-no-lang-sdk-source-display.md), "Why Not" #3). ## References