424 lines
14 KiB
Go
424 lines
14 KiB
Go
// Package bootstrap assembles the proxy-gateway process from data-plane
|
|
// components without putting Controller or storage access on the request path.
|
|
package bootstrap
|
|
|
|
import (
|
|
"bytes"
|
|
"context"
|
|
"errors"
|
|
"fmt"
|
|
"net"
|
|
"reflect"
|
|
"strings"
|
|
"time"
|
|
|
|
"github.com/prometheus/client_golang/prometheus"
|
|
"google.golang.org/grpc"
|
|
"google.golang.org/grpc/credentials"
|
|
|
|
controlplanev1 "proxy-pool/gen/controlplane/v1"
|
|
"proxy-pool/internal/config"
|
|
"proxy-pool/internal/controlplane/clienttransport"
|
|
proxyDomain "proxy-pool/internal/domain/proxy"
|
|
"proxy-pool/internal/domain/workerruntime"
|
|
"proxy-pool/internal/gateway/controlplane"
|
|
"proxy-pool/internal/gateway/dispatch"
|
|
gatewayOutcome "proxy-pool/internal/gateway/outcome"
|
|
"proxy-pool/internal/gateway/server"
|
|
"proxy-pool/internal/gateway/snapshot"
|
|
"proxy-pool/internal/gateway/transport"
|
|
"proxy-pool/internal/platform/httpserver"
|
|
"proxy-pool/internal/platform/lifecycle"
|
|
platformMetrics "proxy-pool/internal/platform/metrics"
|
|
)
|
|
|
|
var (
|
|
ErrInvalidOptions = errors.New("invalid gateway bootstrap options")
|
|
ErrStartup = errors.New("gateway startup failed")
|
|
ErrNotReady = errors.New("gateway snapshot is not ready")
|
|
)
|
|
|
|
const (
|
|
defaultReconnectInitialDelay = time.Second
|
|
defaultReconnectMaxDelay = 30 * time.Second
|
|
defaultReconnectJitter = 20
|
|
defaultOutcomeQueueCapacity = 65_536
|
|
defaultOutcomeBatchSize = 512
|
|
)
|
|
|
|
// Options provides process-local settings. ControlPlaneAddress is deliberately
|
|
// independent from controlPlane.listen: the latter is the Controller bind
|
|
// address, while this value is the Gateway's remote dial target.
|
|
type Options struct {
|
|
ConfigPath string
|
|
Resolver config.Resolver
|
|
ControlPlaneAddress string
|
|
ClusterID string
|
|
WorkerID string
|
|
InstanceID string
|
|
Zone string
|
|
Labels map[string]string
|
|
HTTP httpserver.Options
|
|
|
|
// The pre-bound listeners and transport are test seams. Production leaves
|
|
// them nil and derives listeners and mTLS credentials from configuration.
|
|
GatewayListener net.Listener
|
|
MetricsListener net.Listener
|
|
GRPCTransport credentials.TransportCredentials
|
|
|
|
ReconnectInitialDelay time.Duration
|
|
ReconnectMaxDelay time.Duration
|
|
ReconnectJitter int
|
|
}
|
|
|
|
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.Gateway.Enabled || !configuration.ControlPlane.Enabled {
|
|
return errors.Join(ErrInvalidOptions, errors.New("gateway and controlPlane must be enabled"))
|
|
}
|
|
|
|
runtime, err := newRuntime(ctx, configuration, options)
|
|
if err != nil {
|
|
return fmt.Errorf("%w: %w", ErrStartup, err)
|
|
}
|
|
defer runtime.Close()
|
|
return runtime.Run(ctx)
|
|
}
|
|
|
|
func validateOptions(ctx context.Context, options Options) error {
|
|
if ctx == nil || strings.TrimSpace(options.ConfigPath) != options.ConfigPath || options.ConfigPath == "" ||
|
|
nilInterface(options.Resolver) || !workerruntime.ValidIdentifier(options.ClusterID) || !workerruntime.ValidIdentifier(options.WorkerID) ||
|
|
!workerruntime.ValidIdentifier(options.InstanceID) || !workerruntime.ValidIdentifier(options.Zone) || !clienttransport.ValidDialAddress(options.ControlPlaneAddress) {
|
|
return ErrInvalidOptions
|
|
}
|
|
if options.GatewayListener == nil && options.MetricsListener != nil {
|
|
return ErrInvalidOptions
|
|
}
|
|
if options.ReconnectInitialDelay < 0 || options.ReconnectMaxDelay < 0 || options.ReconnectJitter < 0 || options.ReconnectJitter > 100 {
|
|
return ErrInvalidOptions
|
|
}
|
|
return nil
|
|
}
|
|
|
|
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)
|
|
}
|
|
if err := ctx.Err(); err != nil {
|
|
return nil, 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
|
|
}
|
|
|
|
type runtime struct {
|
|
connection *grpc.ClientConn
|
|
group *lifecycle.Group
|
|
}
|
|
|
|
func newRuntime(ctx context.Context, configuration *config.Config, options Options) (*runtime, error) {
|
|
if ctx == nil || configuration == nil {
|
|
return nil, ErrInvalidOptions
|
|
}
|
|
store := snapshot.NewStore(options.ClusterID, options.WorkerID)
|
|
protection, err := server.BuildProtection(configuration.Gateway)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("build gateway protections: %w", err)
|
|
}
|
|
targets, err := server.TargetPolicyFromListener(configuration.Gateway)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("build gateway target policy: %w", err)
|
|
}
|
|
proxyTransport := transport.New(transport.Config{}, snapshotCredentialResolver{store: store})
|
|
outcomes, err := gatewayOutcome.NewQueue(gatewayOutcome.QueueOptions{
|
|
Capacity: defaultOutcomeQueueCapacity, MaxBatch: min(defaultOutcomeBatchSize, configuration.ControlPlane.MaxRuntimeCounters),
|
|
})
|
|
if err != nil {
|
|
proxyTransport.CloseIdleConnections()
|
|
return nil, fmt.Errorf("build gateway outcome queue: %w", err)
|
|
}
|
|
handler, err := server.New(server.ConfigFromListener(configuration.Gateway), server.Dependencies{
|
|
Auth: protection.Auth,
|
|
Access: protection.Access,
|
|
Admission: protection.Admission,
|
|
Targets: targets,
|
|
Router: server.NewSnapshotRouter(store),
|
|
Dispatcher: dispatch.New(store),
|
|
Transport: proxyTransport,
|
|
Outcomes: outcomes,
|
|
})
|
|
if err != nil {
|
|
proxyTransport.CloseIdleConnections()
|
|
return nil, fmt.Errorf("build gateway handler: %w", err)
|
|
}
|
|
|
|
transportCredentials, err := controlPlaneTransport(configuration.ControlPlane, options.ControlPlaneAddress, options.GRPCTransport)
|
|
if err != nil {
|
|
proxyTransport.CloseIdleConnections()
|
|
return nil, err
|
|
}
|
|
connection, err := grpc.NewClient(options.ControlPlaneAddress,
|
|
grpc.WithTransportCredentials(transportCredentials),
|
|
grpc.WithDefaultCallOptions(
|
|
grpc.MaxCallRecvMsgSize(configuration.ControlPlane.MaxMessageBytes),
|
|
grpc.MaxCallSendMsgSize(configuration.ControlPlane.MaxMessageBytes),
|
|
),
|
|
)
|
|
if err != nil {
|
|
proxyTransport.CloseIdleConnections()
|
|
return nil, fmt.Errorf("dial control plane: %w", err)
|
|
}
|
|
|
|
closeConnection := true
|
|
defer func() {
|
|
if closeConnection {
|
|
_ = connection.Close()
|
|
proxyTransport.CloseIdleConnections()
|
|
}
|
|
}()
|
|
client := controlplanev1.NewWorkerControlPlaneClient(connection)
|
|
reporter, err := controlplane.NewRuntimeReporter(generatedRuntimeClient{client: client}, store, controlplane.Options{
|
|
WorkerID: options.WorkerID, InstanceID: options.InstanceID, Zone: options.Zone,
|
|
ProtocolVersion: configuration.ControlPlane.ProtocolVersion, Labels: options.Labels, Now: time.Now,
|
|
})
|
|
if err != nil {
|
|
return nil, fmt.Errorf("build runtime reporter: %w", err)
|
|
}
|
|
watcher, err := controlplane.NewSnapshotWatcher(controlplane.NewGeneratedSnapshotRPCClient(client), store,
|
|
controlplane.SnapshotWatcherOptions{ClusterID: options.ClusterID, WorkerID: options.WorkerID})
|
|
if err != nil {
|
|
return nil, fmt.Errorf("build snapshot watcher: %w", err)
|
|
}
|
|
outcomeReporter, err := controlplane.NewOutcomeReporter(controlplane.NewGeneratedOutcomeRPCClient(client), outcomes,
|
|
controlplane.OutcomeReporterOptions{WorkerID: options.WorkerID})
|
|
if err != nil {
|
|
return nil, fmt.Errorf("build outcome reporter: %w", err)
|
|
}
|
|
session, err := controlplane.NewSessionRunner(reporter, watcher, outcomeReporter)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("build control plane session: %w", err)
|
|
}
|
|
reconnect, err := controlplane.NewSessionSupervisor(session, reconnectOptions(options))
|
|
if err != nil {
|
|
return nil, fmt.Errorf("build control plane reconnect supervisor: %w", err)
|
|
}
|
|
|
|
httpRuntime, err := newHTTPRuntime(ctx, configuration, options, handler, store)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
group, err := lifecycle.NewGroup(httpRuntime, reconnect)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
closeConnection = false
|
|
return &runtime{connection: connection, group: group}, nil
|
|
}
|
|
|
|
type generatedRuntimeClient struct {
|
|
client controlplanev1.WorkerControlPlaneClient
|
|
}
|
|
|
|
type snapshotCredentialResolver struct {
|
|
store *snapshot.Store
|
|
}
|
|
|
|
func (resolver snapshotCredentialResolver) Resolve(ctx context.Context, selected proxyDomain.Proxy) (transport.Credentials, error) {
|
|
credential, err := resolver.store.Credential(ctx, selected)
|
|
if err != nil {
|
|
return transport.Credentials{}, err
|
|
}
|
|
return transport.Credentials{Username: credential.Username, Password: credential.Password}, nil
|
|
}
|
|
|
|
func (client generatedRuntimeClient) RegisterWorker(
|
|
ctx context.Context,
|
|
request *controlplanev1.RegisterWorkerRequest,
|
|
) (*controlplanev1.RegisterWorkerResponse, error) {
|
|
return client.client.RegisterWorker(ctx, request)
|
|
}
|
|
|
|
func (client generatedRuntimeClient) ReportRuntime(
|
|
ctx context.Context,
|
|
request *controlplanev1.ReportRuntimeRequest,
|
|
) (*controlplanev1.ReportRuntimeResponse, error) {
|
|
return client.client.ReportRuntime(ctx, request)
|
|
}
|
|
|
|
func (runtime *runtime) Run(ctx context.Context) error {
|
|
if runtime == 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()
|
|
}
|
|
}
|
|
|
|
func reconnectOptions(options Options) controlplane.ReconnectOptions {
|
|
initial := options.ReconnectInitialDelay
|
|
if initial == 0 {
|
|
initial = defaultReconnectInitialDelay
|
|
}
|
|
maximum := options.ReconnectMaxDelay
|
|
if maximum == 0 {
|
|
maximum = defaultReconnectMaxDelay
|
|
}
|
|
jitter := options.ReconnectJitter
|
|
if jitter == 0 {
|
|
jitter = defaultReconnectJitter
|
|
}
|
|
return controlplane.ReconnectOptions{InitialDelay: initial, MaxDelay: maximum, Jitter: jitter}
|
|
}
|
|
|
|
type httpRuntime struct {
|
|
options httpserver.Options
|
|
handler *server.Handler
|
|
endpoints []httpserver.Endpoint
|
|
}
|
|
|
|
func newHTTPRuntime(
|
|
ctx context.Context,
|
|
configuration *config.Config,
|
|
options Options,
|
|
handler *server.Handler,
|
|
store *snapshot.Store,
|
|
) (*httpRuntime, error) {
|
|
if ctx == nil || configuration == nil || handler == nil || store == nil {
|
|
return nil, ErrInvalidOptions
|
|
}
|
|
endpoints := make([]httpserver.Endpoint, 0, 2)
|
|
gatewayListener, gatewayOwned, err := resolveListener(ctx, configuration.Gateway.Listen, options.GatewayListener)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("listen gateway: %w", err)
|
|
}
|
|
endpoints = append(endpoints, httpserver.Endpoint{Name: "gateway", Listener: gatewayListener, Handler: handler})
|
|
if configuration.Metrics.Enabled {
|
|
metricsHandler, metricsErr := platformMetrics.NewHandler(platformMetrics.Dependencies{
|
|
Gatherer: prometheus.DefaultGatherer,
|
|
Readiness: snapshotReadiness{store: store, now: time.Now},
|
|
})
|
|
if metricsErr != nil {
|
|
if gatewayOwned {
|
|
_ = gatewayListener.Close()
|
|
}
|
|
return nil, fmt.Errorf("build gateway metrics: %w", metricsErr)
|
|
}
|
|
metricsListener, metricsOwned, listenErr := resolveListener(ctx, configuration.Metrics.Listen, options.MetricsListener)
|
|
if listenErr != nil {
|
|
if gatewayOwned {
|
|
_ = gatewayListener.Close()
|
|
}
|
|
return nil, fmt.Errorf("listen gateway metrics: %w", listenErr)
|
|
}
|
|
_ = metricsOwned
|
|
endpoints = append(endpoints, httpserver.Endpoint{Name: "metrics", Listener: metricsListener, Handler: metricsHandler})
|
|
} else if options.MetricsListener != nil {
|
|
if gatewayOwned {
|
|
_ = gatewayListener.Close()
|
|
}
|
|
return nil, ErrInvalidOptions
|
|
}
|
|
return &httpRuntime{options: options.HTTP, handler: handler, endpoints: endpoints}, nil
|
|
}
|
|
|
|
func (runtime *httpRuntime) Run(ctx context.Context) error {
|
|
if runtime == nil || runtime.handler == nil || len(runtime.endpoints) == 0 || ctx == nil {
|
|
return ErrInvalidOptions
|
|
}
|
|
completed := make(chan struct{})
|
|
go func() {
|
|
select {
|
|
case <-ctx.Done():
|
|
shutdownContext, cancel := context.WithTimeout(context.Background(), shutdownTimeout(runtime.options))
|
|
defer cancel()
|
|
_ = runtime.handler.Shutdown(shutdownContext)
|
|
case <-completed:
|
|
}
|
|
}()
|
|
err := httpserver.Serve(ctx, runtime.options, runtime.endpoints...)
|
|
close(completed)
|
|
return err
|
|
}
|
|
|
|
func resolveListener(ctx context.Context, address string, listener net.Listener) (net.Listener, bool, error) {
|
|
if listener != nil {
|
|
return listener, false, nil
|
|
}
|
|
resolved, err := (&net.ListenConfig{}).Listen(ctx, "tcp", address)
|
|
if err != nil {
|
|
return nil, false, err
|
|
}
|
|
return resolved, true, nil
|
|
}
|
|
|
|
func shutdownTimeout(options httpserver.Options) time.Duration {
|
|
if options.ShutdownTimeout > 0 {
|
|
return options.ShutdownTimeout
|
|
}
|
|
return httpserver.DefaultOptions().ShutdownTimeout
|
|
}
|
|
|
|
type snapshotReadiness struct {
|
|
store *snapshot.Store
|
|
now func() time.Time
|
|
}
|
|
|
|
func (readiness snapshotReadiness) Ready(ctx context.Context) error {
|
|
if ctx == nil || readiness.store == nil || readiness.now == nil {
|
|
return ErrNotReady
|
|
}
|
|
if err := ctx.Err(); err != nil {
|
|
return err
|
|
}
|
|
current := readiness.store.Current()
|
|
if current == nil || current.ValidUntil.IsZero() || !current.ValidUntil.After(readiness.now().UTC()) {
|
|
return ErrNotReady
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func controlPlaneTransport(
|
|
configuration config.ControlPlane,
|
|
address string,
|
|
override credentials.TransportCredentials,
|
|
) (credentials.TransportCredentials, error) {
|
|
transport, err := clienttransport.New(configuration, address, configuration.GatewayTLS, override, "gatewayTLS")
|
|
if err != nil {
|
|
return nil, fmt.Errorf("%w: %v", ErrInvalidOptions, err)
|
|
}
|
|
return transport, 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
|
|
}
|
|
}
|