File tree Expand file tree Collapse file tree
Expand file tree Collapse file tree Original file line number Diff line number Diff line change @@ -195,7 +195,7 @@ func (m *taskHandlerMap) ListTaskHandlers() []bundle.TaskHandlerInfo {
195195 return slices .Clone (m .order )
196196}
197197
198- // dagMap holds the registered Dags by dag_id and keeps the order in which they were registered .
198+ // dagMap holds the registered Dags by dag_id, in registration order .
199199type dagMap struct {
200200 mu sync.Mutex
201201 dags map [string ]* DagRef
@@ -225,9 +225,8 @@ func (m *dagMap) has(dagID string) bool {
225225 return exists
226226}
227227
228- // serialize serializes the registered Dags in the order they were registered. If serializing a Dag
229- // panics, the panic becomes the error of that Dag instead of ending the parse. The import error of
230- // the parse then names the Dag.
228+ // serialize serializes the Dags in registration order. If a Dag panics, the panic becomes that
229+ // Dag's Err.
231230func (m * dagMap ) serialize (fileloc , relativeFileloc string ) []bundle.SerializedDag {
232231 m .mu .Lock ()
233232 dags := slices .Clone (m .order )
Original file line number Diff line number Diff line change @@ -418,8 +418,7 @@ func TestSerializeDagsKeepsTheOtherDagsWhenADagCannotBeSerialized(t *testing.T)
418418 etl := Dag ("etl" )
419419 etl .Task (noop )
420420 b .Register (etl )
421- // Register checks and expands every Dag it takes. A Dag that skipped Register therefore stands
422- // in for a Dag that the serializer fails on.
421+ // A Dag that skipped Register stands in for one the serializer fails on.
423422 broken := Dag ("broken" )
424423 b .dags .dags ["broken" ] = broken
425424 b .dags .order = append (b .dags .order , broken )
Original file line number Diff line number Diff line change @@ -71,9 +71,8 @@ type DagRef struct {
7171// [IfRef.Then], [IfRef.Else], [SwitchRef.Case] and the methods of [TaskGroupRef] panic once the Dag
7272// is registered.
7373//
74- // When the Dag processor asks to parse the bundle, [BundleRef.Serve] answers with the registered
75- // Dags in serialized form. Serve does not yet list these Dags in the --airflow-metadata manifest or
76- // run their tasks.
74+ // [BundleRef.Serve] sends the registered Dags to the Dag processor, but does not yet list them in
75+ // the --airflow-metadata manifest or run their tasks.
7776//
7877// Dag panics if it gets more than one DagSpec, or if the DagSpec has a value that Python rejects
7978// when it builds or validates a Dag:
Original file line number Diff line number Diff line change @@ -57,10 +57,9 @@ const (
5757// The command-line flags of the executable decide what Serve does.
5858// With --airflow-metadata it prints the bundle's manifest and returns, which is how
5959// airflow-go-pack reads the Dag and task ids of the registered task handlers.
60- // Airflow starts the executable with --comm and --logs. With these flags, Serve speaks the
61- // coordinator protocol. It either runs one task or answers the Dag processor's request to parse
62- // the bundle. To answer a parse request, Serve sends the Dags from [Dag] that the bundle registered
63- // and runs no task.
60+ // Airflow starts the executable with --comm and --logs. Serve then speaks the coordinator
61+ // protocol and either runs one task or answers the Dag processor's parse request with the Dags
62+ // from [Dag].
6463//
6564// main must exit with a non-zero status when Serve returns an error, because the exit status
6665// is how the supervisor learns that the task failed:
Original file line number Diff line number Diff line change @@ -220,9 +220,8 @@ func TestServeHelpIsNotAnError(t *testing.T) {
220220 assert .Contains (t , stdout .String (), "--airflow-metadata" )
221221}
222222
223- // serveForSupervisor runs b.serve with --comm and --logs, the same flags that Airflow starts the
224- // bundle with. It returns the supervisor's end of the comm socket and the channel that gets what
225- // serve returns.
223+ // serveForSupervisor runs b.serve with --comm and --logs and returns the supervisor's end of the
224+ // comm socket and the channel that gets what serve returns.
226225func serveForSupervisor (t * testing.T , b * BundleRef ) (* execution.CoordinatorComm , <- chan error ) {
227226 t .Helper ()
228227 commLn , err := net .Listen ("tcp" , "127.0.0.1:0" )
@@ -294,8 +293,7 @@ func TestServeRunsTaskForSupervisor(t *testing.T) {
294293 assert .True (t , ran )
295294}
296295
297- // To parse the bundle, the Dag processor starts it with --comm and --logs and then sends a
298- // DagFileParseRequest over the comm socket.
296+ // The Dag processor sends a DagFileParseRequest instead of StartupDetails.
299297func TestServeAnswersTheDagParseRequestWithoutRunningATask (t * testing.T ) {
300298 var ran []string
301299 record := func (name string ) func (Context ) error {
Original file line number Diff line number Diff line change 1515// specific language governing permissions and limitations
1616// under the License.
1717
18- // Package bundle defines what the coordinator runtime needs from a bundle. The
19- // runtime needs the tasks to look up and run, and the Dag and task ids to list
20- // in the manifest. It also needs the serialized Dags to send in answer to a Dag
21- // parse request.
18+ // Package bundle defines what the coordinator runtime needs from a bundle: the
19+ // tasks it looks up and runs, the Dag and task ids it lists in the manifest, and
20+ // the serialized Dags it sends to the Dag processor.
2221//
23- // Package airflow builds the tasks, the ids, and the serialized Dags from the
24- // task handlers and the Dags that a bundle registers .
22+ // Package airflow builds the tasks and ids from the task handlers a bundle
23+ // registers, and the serialized Dags from its Dags .
2524package bundle
Original file line number Diff line number Diff line change @@ -33,8 +33,7 @@ import (
3333 "github.com/apache/airflow/go-sdk/pkg/execution/genmodels"
3434)
3535
36- // serializedDags serializes the Dags it holds. It also records the fileloc and relativeFileloc
37- // that SerializeDags was called with.
36+ // serializedDags is a DagSerializer that records the paths it was called with.
3837type serializedDags struct {
3938 dags []bundle.SerializedDag
4039 fileloc , relative string
@@ -60,8 +59,7 @@ var etlParseRequest = &genmodels.DagFileParseRequest{
6059
6160func discardLogger () * slog.Logger { return slog .New (slog .NewTextHandler (io .Discard , nil )) }
6261
63- // wireJSON passes body through the frame encoding that SendRequest uses and returns the result as
64- // JSON. A test then sees the keys, nulls and empty lists that the Dag processor receives.
62+ // wireJSON returns body as the Dag processor receives it, after the frame encoding of SendRequest.
6563func wireJSON (t * testing.T , body any ) string {
6664 t .Helper ()
6765 payload , err := encodeRequest (0 , body )
Original file line number Diff line number Diff line change @@ -1361,8 +1361,8 @@ func TestServeFailureAfterConnectClosesComm(t *testing.T) {
13611361 logsConn := <- logsCh
13621362 defer logsConn .Close ()
13631363
1364- // Serve expects StartupDetails or DagFileParseRequest as the first frame , so it fails to
1365- // decode a VariableResult.
1364+ // Serve expects StartupDetails or DagFileParseRequest first, so it fails to decode a
1365+ // VariableResult.
13661366 payload , err := encodeRequest (
13671367 0 ,
13681368 map [string ]any {"type" : "VariableResult" , "key" : "k" , "value" : "v" },
Original file line number Diff line number Diff line change 2020// the Airflow supervisor (Python ExecutableCoordinator), the Serve method of
2121// airflow.BundleRef dispatches here.
2222//
23- // The first frame that the runtime reads on the comm socket decides what the
24- // runtime does. After a StartupDetails message, the runtime runs a task over
25- // several rounds of messages. After a DagFileParseRequest from the Dag
26- // processor, the runtime runs no task. It answers with one DagFileParsingResult
27- // that holds the Dags from airflow.Dag that the bundle registered.
23+ // The first frame on the comm socket picks the mode: StartupDetails runs one
24+ // task, and DagFileParseRequest is answered with the bundle's serialized Dags.
2825//
2926// See go-sdk/adr/0003-coordinator-protocol-msgpack-ipc.md.
3027package execution
You can’t perform that action at this time.
0 commit comments