// Package bootstrap assembles the standalone proxy-checker process. Checker // execution is intentionally isolated from Gateway request handling and from // all Redis/PostgreSQL access. package bootstrap import ( "bytes" "context" "errors" "fmt" "net" "net/http" "reflect" "strings" "sync/atomic" "github.com/prometheus/client_golang/prometheus" "google.golang.org/grpc" "google.golang.org/grpc/credentials" controlplanev1 "proxy-pool/gen/controlplane/v1" "proxy-pool/internal/checker/controlplane" "proxy-pool/internal/checker/probe" "proxy-pool/internal/config" "proxy-pool/internal/controlplane/clienttransport" "proxy-pool/internal/controlplane/tlsreload" "proxy-pool/internal/domain/workerruntime" "proxy-pool/internal/platform/httpserver" "proxy-pool/internal/platform/lifecycle" platformMetrics "proxy-pool/internal/platform/metrics" ) var ( ErrInvalidOptions = errors.New("invalid checker bootstrap options") ErrStartup = errors.New("checker startup failed") ErrNotReady = errors.New("checker control plane is not ready") ) type Options struct { ConfigPath string Resolver config.Resolver ControlPlaneAddress string CheckerID string InstanceID string AutoIdentity bool MaxInFlight int SupportedLevels []controlplanev1.CheckLevel GRPCTransport credentials.TransportCredentials MetricsListener net.Listener HTTP httpserver.Options } func Run(ctx context.Context, options Options) error { if err := validateOptions(ctx, options); err != nil { return err } configuration, err := loadConfiguration(ctx, options.ConfigPath, options.Resolver) if err != nil { return fmt.Errorf("%w: load configuration: %w", ErrStartup, err) } if !configuration.ControlPlane.Enabled { return errors.Join(ErrInvalidOptions, errors.New("controlPlane must be enabled")) } options, err = resolveIdentity(configuration, options) if err != nil { return fmt.Errorf("%w: %w", ErrStartup, err) } runtime, err := newRuntime(ctx, configuration, options) if err != nil { return fmt.Errorf("%w: %w", ErrStartup, err) } defer runtime.Close() return runtime.Run(ctx) } func newRuntime(ctx context.Context, configuration *config.Config, options Options) (*runtime, error) { if ctx == nil || configuration == nil { return nil, ErrInvalidOptions } transport, err := clienttransport.New( configuration.ControlPlane, options.ControlPlaneAddress, configuration.ControlPlane.CheckerTLS, options.GRPCTransport, "checkerTLS", ) if err != nil { return nil, fmt.Errorf("%w: %w", ErrInvalidOptions, err) } connection, err := grpc.NewClient(options.ControlPlaneAddress, grpc.WithTransportCredentials(transport), grpc.WithDefaultCallOptions( grpc.MaxCallRecvMsgSize(configuration.ControlPlane.MaxMessageBytes), grpc.MaxCallSendMsgSize(configuration.ControlPlane.MaxMessageBytes), ), ) if err != nil { return nil, fmt.Errorf("dial control plane: %w", err) } closeConnection := true defer func() { if closeConnection { _ = connection.Close() } }() client := controlplane.NewGeneratedClient(controlplanev1.NewCheckerControlPlaneClient(connection)) readiness := &controlPlaneReadiness{} runner, err := controlplane.NewRunner(client, probe.NewExecutor(), controlplane.Options{ CheckerID: options.CheckerID, InstanceID: options.InstanceID, MaxInFlight: options.MaxInFlight, SupportedLevels: append([]controlplanev1.CheckLevel(nil), options.SupportedLevels...), ReportBatchSize: options.MaxInFlight, OnSuccessfulPull: readiness.MarkReady, OnFailedPull: readiness.MarkUnavailable, }) if err != nil { return nil, fmt.Errorf("build checker runner: %w", err) } runners := []lifecycle.Runner{runner} if configuration.Metrics.Enabled { metrics, metricsErr := newMetricsRuntime(ctx, configuration.Metrics, options, readiness) if metricsErr != nil { return nil, metricsErr } runners = append(runners, metrics) } else if options.MetricsListener != nil { return nil, ErrInvalidOptions } group, err := lifecycle.NewGroup(runners...) if err != nil { return nil, err } closeConnection = false return &runtime{connection: connection, group: group}, nil } func validateOptions(ctx context.Context, options Options) error { if ctx == nil || strings.TrimSpace(options.ConfigPath) != options.ConfigPath || options.ConfigPath == "" || nilInterface(options.Resolver) || !clienttransport.ValidDialAddress(options.ControlPlaneAddress) || (options.CheckerID != "" && !workerruntime.ValidIdentifier(options.CheckerID)) || (options.InstanceID != "" && !workerruntime.ValidIdentifier(options.InstanceID)) || (!options.AutoIdentity && (!workerruntime.ValidIdentifier(options.CheckerID) || !workerruntime.ValidIdentifier(options.InstanceID))) || options.MaxInFlight <= 0 || len(options.SupportedLevels) == 0 { return ErrInvalidOptions } return nil } func resolveIdentity(configuration *config.Config, options Options) (Options, error) { if configuration == nil { return Options{}, ErrInvalidOptions } if !options.AutoIdentity { return options, nil } if configuration.ControlPlane.TLS.Mode != "mtls" { return Options{}, errors.Join(ErrInvalidOptions, errors.New("auto identity requires controlPlane mTLS")) } identity, err := tlsreload.ResolveSPIFFEIdentity( configuration.ControlPlane.CheckerTLS.CertFile, configuration.ControlPlane.CheckerTLS.KeyFile, configuration.ControlPlane.TLS.TrustDomain, configuration.ControlPlane.TLS.Environment, "checker", ) if err != nil { return Options{}, fmt.Errorf("resolve checker SPIFFE identity: %w", err) } if options.CheckerID == "" { options.CheckerID = identity } if options.InstanceID == "" { options.InstanceID = identity } if !workerruntime.ValidIdentifier(options.CheckerID) || !workerruntime.ValidIdentifier(options.InstanceID) || options.CheckerID != identity { return Options{}, errors.Join(ErrInvalidOptions, errors.New("checker identity does not match SPIFFE certificate")) } return options, nil } type runtime struct { connection *grpc.ClientConn group *lifecycle.Group } func (runtime *runtime) Run(ctx context.Context) error { if runtime == nil || runtime.connection == nil || runtime.group == nil || ctx == nil { return ErrInvalidOptions } return runtime.group.Run(ctx) } func (runtime *runtime) Close() { if runtime != nil && runtime.connection != nil { _ = runtime.connection.Close() } } type controlPlaneReadiness struct { ready atomic.Bool } func (readiness *controlPlaneReadiness) MarkReady() { if readiness != nil { readiness.ready.Store(true) } } func (readiness *controlPlaneReadiness) MarkUnavailable() { if readiness != nil { readiness.ready.Store(false) } } func (readiness *controlPlaneReadiness) Ready(ctx context.Context) error { if ctx == nil || readiness == nil || !readiness.ready.Load() { return ErrNotReady } return ctx.Err() } type metricsRuntime struct { listener net.Listener handler http.Handler options httpserver.Options } func newMetricsRuntime( ctx context.Context, configuration config.Metrics, options Options, readiness *controlPlaneReadiness, ) (*metricsRuntime, error) { if ctx == nil || !configuration.Enabled || readiness == nil { return nil, ErrInvalidOptions } handler, err := platformMetrics.NewHandler(platformMetrics.Dependencies{ Gatherer: prometheus.DefaultGatherer, Readiness: readiness, }) if err != nil { return nil, fmt.Errorf("build checker metrics handler: %w", err) } listener := options.MetricsListener if listener == nil { listener, err = (&net.ListenConfig{}).Listen(ctx, "tcp", configuration.Listen) if err != nil { return nil, fmt.Errorf("listen checker metrics: %w", err) } } return &metricsRuntime{listener: listener, handler: handler, options: options.HTTP}, nil } func (runtime *metricsRuntime) Run(ctx context.Context) error { if runtime == nil || runtime.listener == nil || runtime.handler == nil || ctx == nil { return ErrInvalidOptions } return httpserver.Serve(ctx, runtime.options, httpserver.Endpoint{ Name: "metrics", Listener: runtime.listener, Handler: runtime.handler, }) } func loadConfiguration(ctx context.Context, path string, resolver config.Resolver) (*config.Config, error) { if ctx == nil || strings.TrimSpace(path) != path || path == "" || nilInterface(resolver) { return nil, ErrInvalidOptions } if err := ctx.Err(); err != nil { return nil, err } content, err := resolver.ReadFile(path) if err != nil { return nil, fmt.Errorf("read configuration %q: %w", path, err) } configuration, err := config.LoadResolved(bytes.NewReader(content), resolver) if err != nil { return nil, fmt.Errorf("decode configuration %q: %w", path, err) } return configuration, nil } func nilInterface(value any) bool { if value == nil { return true } reflected := reflect.ValueOf(value) switch reflected.Kind() { case reflect.Chan, reflect.Func, reflect.Interface, reflect.Map, reflect.Pointer, reflect.Slice: return reflected.IsNil() default: return false } }