From e181e2c035c393acd951cd481b19d7b6cf490508 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Sat, 25 Apr 2026 13:40:45 -0700 Subject: [PATCH] parquet_pushdown(M0-C3): daemon bootstrap (config, filer client, server) Add weed/parquet_pushdown/daemon, the host process for the SeaweedParquetPushdown service: - Config carries bind ip/port, filer addresses, trust mode, and the ConnectorTrustedAck flag that gates the dev-only connector-trusted trust mode (default refuses it; explicit ack required). - filerClient probes the filer with GetFilerConfiguration at startup so bad endpoints fail fast, mirroring the iam command's pattern. Bounded by a 60s timeout; later milestones use the same client to read Parquet data and side-index blobs. - server.go wires it together: load TLS, probe filer, build the Service, listen on (host, localhost) gRPC ports, register the service + reflection on each, install grace.OnInterrupt for GracefulStop, and Serve. Run() returns an error so the calling weed pushdown command can report startup failures instead of glog.Fatalf. --- weed/parquet_pushdown/daemon/config.go | 66 +++++++++++++++++ weed/parquet_pushdown/daemon/filer_client.go | 48 ++++++++++++ weed/parquet_pushdown/daemon/server.go | 77 ++++++++++++++++++++ 3 files changed, 191 insertions(+) create mode 100644 weed/parquet_pushdown/daemon/config.go create mode 100644 weed/parquet_pushdown/daemon/filer_client.go create mode 100644 weed/parquet_pushdown/daemon/server.go diff --git a/weed/parquet_pushdown/daemon/config.go b/weed/parquet_pushdown/daemon/config.go new file mode 100644 index 000000000..53abcd163 --- /dev/null +++ b/weed/parquet_pushdown/daemon/config.go @@ -0,0 +1,66 @@ +// Package daemon hosts the standalone `weed pushdown` process. +// It owns gRPC server bootstrap, the filer client used by later +// milestones to read Parquet data and side-index blobs, and the +// process-level config plumbing. +package daemon + +import ( + "errors" + "fmt" + + "github.com/seaweedfs/seaweedfs/weed/parquet_pushdown" + "github.com/seaweedfs/seaweedfs/weed/pb" +) + +// Config controls how the daemon listens, talks to the filer, and +// validates incoming requests. Caller (the `weed pushdown` command) +// builds this from CLI flags. +type Config struct { + // IP is the bind address for the gRPC listener. Empty defaults to + // the auto-detected host address. + IP string + + // Port is the gRPC listener port. + Port int + + // FilerAddresses is the comma-resolved list of filer endpoints + // the daemon talks to for data reads and (in M3) catalog access. + // Required: at least one entry. + FilerAddresses []pb.ServerAddress + + // TrustMode is the request-validation strictness. Default is + // "catalog-validated"; "connector-trusted" is a developer-only + // mode that the daemon refuses to accept in production builds + // (see ConnectorTrustedAck). + TrustMode parquet_pushdown.TrustMode + + // ConnectorTrustedAck is the explicit acknowledgement required to + // run with TrustMode == TrustModeConnectorTrusted. The flag exists + // to make the dev-only mode awkward to enable by accident. + ConnectorTrustedAck bool + + // Version is reported in PingResponse / PushdownStats. Caller + // usually fills this with weed/util/version.Version(). + Version string +} + +// Validate checks the config before the daemon starts. +func (c *Config) Validate() error { + if c.Port <= 0 || c.Port > 65535 { + return fmt.Errorf("port %d is out of range", c.Port) + } + if len(c.FilerAddresses) == 0 { + return errors.New("at least one filer address is required") + } + switch c.TrustMode { + case "", parquet_pushdown.TrustModeCatalogValidated: + c.TrustMode = parquet_pushdown.TrustModeCatalogValidated + case parquet_pushdown.TrustModeConnectorTrusted: + if !c.ConnectorTrustedAck { + return errors.New("connector-trusted is a developer-only trust mode; pass -dev.connector_trusted_ack to enable") + } + default: + return fmt.Errorf("unknown trust mode %q", c.TrustMode) + } + return nil +} diff --git a/weed/parquet_pushdown/daemon/filer_client.go b/weed/parquet_pushdown/daemon/filer_client.go new file mode 100644 index 000000000..d42ad1ae8 --- /dev/null +++ b/weed/parquet_pushdown/daemon/filer_client.go @@ -0,0 +1,48 @@ +package daemon + +import ( + "context" + "fmt" + "time" + + "google.golang.org/grpc" + + "github.com/seaweedfs/seaweedfs/weed/glog" + "github.com/seaweedfs/seaweedfs/weed/pb" + "github.com/seaweedfs/seaweedfs/weed/pb/filer_pb" +) + +// filerClient is a thin handle around the daemon's filer endpoints. +// M0 only uses it to confirm filer reachability at startup; later +// milestones add read paths for Parquet data and side-index blobs. +type filerClient struct { + addresses []pb.ServerAddress + grpcDialOption grpc.DialOption +} + +func newFilerClient(addresses []pb.ServerAddress, grpcDialOption grpc.DialOption) *filerClient { + return &filerClient{addresses: addresses, grpcDialOption: grpcDialOption} +} + +// waitUntilReachable blocks until the daemon can complete a +// GetFilerConfiguration RPC against any of its configured filer +// addresses, or the context is cancelled. Mirrors the iam command's +// startup-time filer probe. +func (c *filerClient) waitUntilReachable(ctx context.Context) error { + for { + err := pb.WithOneOfGrpcFilerClients(false, c.addresses, c.grpcDialOption, func(client filer_pb.SeaweedFilerClient) error { + _, getErr := client.GetFilerConfiguration(ctx, &filer_pb.GetFilerConfigurationRequest{}) + return getErr + }) + if err == nil { + glog.V(0).Infof("pushdown daemon connected to filers %v", c.addresses) + return nil + } + glog.V(0).Infof("pushdown daemon waiting for filers %v: %v", c.addresses, err) + select { + case <-ctx.Done(): + return fmt.Errorf("filer reachability wait cancelled: %w", ctx.Err()) + case <-time.After(time.Second): + } + } +} diff --git a/weed/parquet_pushdown/daemon/server.go b/weed/parquet_pushdown/daemon/server.go new file mode 100644 index 000000000..4b22d8095 --- /dev/null +++ b/weed/parquet_pushdown/daemon/server.go @@ -0,0 +1,77 @@ +package daemon + +import ( + "context" + "fmt" + "net" + "time" + + "google.golang.org/grpc" + "google.golang.org/grpc/reflection" + + "github.com/seaweedfs/seaweedfs/weed/glog" + "github.com/seaweedfs/seaweedfs/weed/parquet_pushdown" + "github.com/seaweedfs/seaweedfs/weed/pb" + pbpushdown "github.com/seaweedfs/seaweedfs/weed/pb/parquet_pushdown_pb" + "github.com/seaweedfs/seaweedfs/weed/security" + "github.com/seaweedfs/seaweedfs/weed/util" + "github.com/seaweedfs/seaweedfs/weed/util/grace" +) + +// Run starts the pushdown daemon. It blocks until the gRPC server +// returns (graceful shutdown on SIGTERM, or error from Serve). Always +// validates the config first. +func Run(cfg Config) error { + if err := cfg.Validate(); err != nil { + return fmt.Errorf("pushdown config: %w", err) + } + + util.LoadSecurityConfiguration() + grpcDialOption := security.LoadClientTLS(util.GetViper(), "grpc.client") + + // Probe the filer until reachable so we fail fast on bad config + // and emit a clear "waiting for filers" log otherwise. Bound the + // probe so the daemon doesn't sit forever if the operator hands + // in an unreachable filer. + probeCtx, probeCancel := context.WithTimeout(context.Background(), 60*time.Second) + defer probeCancel() + fc := newFilerClient(cfg.FilerAddresses, grpcDialOption) + if err := fc.waitUntilReachable(probeCtx); err != nil { + return fmt.Errorf("filer probe failed: %w", err) + } + + svc := parquet_pushdown.New(parquet_pushdown.Options{ + Version: cfg.Version, + TrustMode: cfg.TrustMode, + }) + + grpcL, localL, err := util.NewIpAndLocalListeners(cfg.IP, cfg.Port, 0) + if err != nil { + return fmt.Errorf("listen on grpc port %d: %w", cfg.Port, err) + } + + grpcS := pb.NewGrpcServer() + pbpushdown.RegisterSeaweedParquetPushdownServer(grpcS, svc) + reflection.Register(grpcS) + + grace.OnInterrupt(grpcS.GracefulStop) + + if localL != nil { + localGrpcS := pb.NewGrpcServer() + pbpushdown.RegisterSeaweedParquetPushdownServer(localGrpcS, svc) + reflection.Register(localGrpcS) + grace.OnInterrupt(localGrpcS.GracefulStop) + go serve(localGrpcS, localL, "pushdown localhost") + } + + glog.V(0).Infof("pushdown daemon listening on %s:%d (trust=%s)", cfg.IP, cfg.Port, cfg.TrustMode) + return serve(grpcS, grpcL, "pushdown") +} + +func serve(s *grpc.Server, l net.Listener, label string) error { + if err := s.Serve(l); err != nil && err != grpc.ErrServerStopped { + glog.Errorf("%s server serve error: %v", label, err) + return err + } + return nil +}