FrankYang0529 commented on code in PR #73426:
URL: https://github.com/apache/airflow/pull/73426#discussion_r4062119764
##########
go-sdk/airflow/serve.go:
##########
@@ -15,113 +15,97 @@
// specific language governing permissions and limitations
// under the License.
-package bundlev1server
+package airflow
import (
"errors"
+ "io"
+ "os"
flag "github.com/spf13/pflag"
- "github.com/apache/airflow/go-sdk/bundle/bundlev1"
"github.com/apache/airflow/go-sdk/pkg/execution"
)
-// ErrCoordinatorFlagsRequired is returned by [Serve] unless both --comm and
+// errCoordinatorFlagsRequired is returned by Serve unless both --comm and
// --logs are supplied. Bundle execution always uses the coordinator protocol.
-var ErrCoordinatorFlagsRequired = errors.New(
+var errCoordinatorFlagsRequired = errors.New(
"--comm and --logs are required for bundle execution",
)
-// ErrFormatRequiresMetadata is returned by [Serve] when --format is supplied
+// errFormatRequiresMetadata is returned by Serve when --format is supplied
// without --airflow-metadata, the only mode whose encoding it selects.
-var ErrFormatRequiresMetadata = errors.New(
+var errFormatRequiresMetadata = errors.New(
"--format is only valid together with --airflow-metadata",
)
-// CLI Flags, all read by Serve to choose a server mode below.
-// --airflow-metadata prints the bundle's manifest and exits (airflow-go-pack
-// consumes it to build the embedded airflow-metadata.yaml); --format selects
-// its encoding. --comm and --logs select coordinator mode.
-var (
- printMetadata = flag.Bool(
+// serveMode tags the protocol the binary will speak this run.
+type serveMode int
+
+const (
+ modeAirflowMetadata serveMode = iota // --airflow-metadata: print
the manifest JSON (ADR 0002/0004)
+ modeCoordinator // --comm/--logs:
msgpack-over-IPC (ADR 0003)
+ modeCoordinatorUsageError // missing coordinator flags
+)
+
+// Serve runs the bundle. Call it as the last statement of main.
+//
+// The command-line flags of the executable decide what Serve does.
+// With --airflow-metadata it prints the bundle's manifest and returns, which
is how
+// airflow-go-pack reads the registered Dag and task ids.
+// With --comm and --logs, which the Airflow supervisor passes, it runs one
task over the
+// coordinator protocol.
+//
+// main must exit with a non-zero status when Serve returns an error, because
the exit status
+// is how the supervisor learns that the task failed:
+//
+// if err := bundle.Serve(); err != nil {
+// log.Fatal(err)
+// }
+func (b *BundleRef) Serve() error {
+ return b.serve(os.Args[1:], os.Stdout)
+}
+
+func (b *BundleRef) serve(args []string, stdout io.Writer) error {
+ // The flags go on their own FlagSet. On pflag.CommandLine, every
program that imports this
+ // package would get them, and one that defines its own --format there
would panic.
+ flags := flag.NewFlagSet(os.Args[0], flag.ContinueOnError)
+ printMetadata := flags.Bool(
"airflow-metadata",
false,
"print the bundle's airflow-metadata manifest and exit",
)
- metadataFormat = flag.String(
+ metadataFormat := flags.String(
"format",
string(execution.MetadataFormatYAML),
"encoding for --airflow-metadata: yaml (default) or json; only
valid with --airflow-metadata",
)
- commAddr = flag.String(
+ commAddr := flags.String(
"comm",
"",
"host:port of the supervisor's coordinator comm channel
(selects coordinator mode)",
)
- logsAddr = flag.String(
+ logsAddr := flags.String(
"logs",
"",
"host:port of the supervisor's coordinator logs channel
(selects coordinator mode)",
)
-)
-
-// ServeOpt is an interface for defining options that can be passed to the
-// Serve function. Each implementation modifies the ServeConfig being
-// generated. A slice of ServeOpts then, cumulatively applied, render a full
-// ServeConfig.
-type ServeOpt interface {
- ApplyServeOpt(*ServerConfig) error
-}
-
-type serveConfigFunc func(*ServerConfig) error
-
-func (s serveConfigFunc) ApplyServeOpt(in *ServerConfig) error {
- return s(in)
-}
-
-// ServerConfig holds settings that ServeOpt values apply before the bundle
-// server starts. It is currently empty; it exists so options can be added
later
-// without changing Serve's signature.
-type ServerConfig struct{}
-
-// serveMode tags the protocol the binary will speak this run.
-type serveMode int
-
-const (
- modeAirflowMetadata serveMode = iota // --airflow-metadata: print
the manifest JSON (ADR 0002/0004)
- modeCoordinator // --comm/--logs:
msgpack-over-IPC (ADR 0003)
- modeCoordinatorUsageError // missing coordinator flags
-)
-
-// Serve is the entrypoint for a bundle executed by Airflow's coordinator.
-//
-// The mode is decided from CLI flags. Callers should
-// surface the returned error so misuse (e.g. only one of --comm/--logs
-// supplied) produces a non-zero exit:
-//
-// func main() {
-// if err := bundlev1server.Serve(&myBundle{}); err != nil {
-// log.Fatal(err)
-// }
-// }
-//
-// Zero or more options to configure the server may also be passed. There are
-// no options yet; the parameter exists to allow future additions without
-// breaking compatibility.
-func Serve(bundle bundlev1.BundleProvider, opts ...ServeOpt) error {
- flag.Parse()
-
- serveConfig := &ServerConfig{}
- for _, c := range opts {
- c.ApplyServeOpt(serveConfig)
+ // A bundle may define flags of its own on pflag.CommandLine. Serve
parses the whole command
+ // line, so it has to accept those too.
+ flags.AddFlagSet(flag.CommandLine)
Review Comment:
Agree, changed to use `flags.VisitAll` instead of listing the four names, so
a flag that `Serve` gains later is covered too.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
