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

@@ -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
+}

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):
+		}
+	}
+}

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
+}