config.go
Rama master · réplica de lectura del repositorio original.
package aastro
import (
"errors"
"fmt"
"io"
"os"
"reflect"
"regexp"
"strings"
"time"
"github.com/creasty/defaults"
"github.com/go-playground/validator/v10"
"gopkg.in/yaml.v3"
)
type Config struct {
Schema string `yaml:"schema" validate:"required,oneof=v1"`
Debug bool `yaml:"debug"`
Gateway GatewayConfig `yaml:"gateway" validate:"required"`
}
func (c *Config) Marshal() ([]byte, error) {
return yaml.Marshal(c)
}
func (c *Config) WriteTo(w io.Writer) (int64, error) {
data, err := c.Marshal()
if err != nil {
return 0, fmt.Errorf("marshal config: %w", err)
}
n, err := w.Write(data)
return int64(n), err
}
type GatewayConfig struct {
Service ServiceConfig `yaml:"service"`
Server ServerConfig `yaml:"server" validate:"required"`
Admin AdminConfig `yaml:"admin" validate:"required"`
Observability ObservabilityConfig `yaml:"observability"`
Routing RoutingConfig `yaml:"routing" validate:"required"`
}
type ServiceConfig struct {
Name string `yaml:"name" default:"aastro"`
}
type ServerConfig struct {
Port int `yaml:"port" validate:"required,min=1,max=65535"`
Timeout time.Duration `yaml:"timeout" default:"5s"`
HeaderTimeout time.Duration `yaml:"header_timeout" default:"5s"`
HTTP2 string `yaml:"http2" default:"auto" validate:"omitempty,oneof=auto on off"`
TLS ServerTLSConfig `yaml:"tls"`
}
type AdminConfig struct {
Port int `yaml:"port" validate:"required,min=1,max=65535"`
BindAddr string `yaml:"bind_addr" default:"127.0.0.1"`
Timeout time.Duration `yaml:"timeout" default:"5m"`
HeaderTimeout time.Duration `yaml:"header_timeout" default:"5s"`
EnablePprof bool `yaml:"enable_pprof" default:"false"`
}
type ObservabilityConfig struct {
Tracing TracingConfig `yaml:"tracing"`
Metrics MetricsConfig `yaml:"metrics"`
}
type MetricsConfig struct {
Enabled bool `yaml:"enabled"`
Exporter string `yaml:"exporter" validate:"required_if=Enabled true,omitempty,oneof=otlp prometheus"`
OTLP OTLPConfig `yaml:"otlp"`
}
type TracingConfig struct {
Enabled bool `yaml:"enabled"`
Exporter string `yaml:"exporter" validate:"required_if=Enabled true,omitempty,oneof=otlp"`
SamplingRatio float64 `yaml:"sampling_ratio" default:"1.0" validate:"min=0,max=1"`
OTLP OTLPConfig `yaml:"otlp"`
}
type OTLPConfig struct {
Endpoint string `yaml:"endpoint"`
Insecure bool `yaml:"insecure"`
Interval time.Duration `yaml:"interval"`
}
type ServerTLSConfig struct {
Enabled bool `yaml:"enabled"`
CertFile string `yaml:"cert_file" validate:"required_if=Enabled true"`
KeyFile string `yaml:"key_file" validate:"required_if=Enabled true"`
MinVersion string `yaml:"min_version" default:"1.2" validate:"omitempty,oneof=1.2 1.3"`
ClientAuth string `yaml:"client_auth" default:"none" validate:"omitempty,oneof=require optional none"`
ClientCAFile string `yaml:"client_ca_file" validate:"required_unless=ClientAuth none"`
}
type RoutingConfig struct {
RateLimiter RateLimiterConfig `yaml:"rate_limiter" validate:"omitempty"`
TrustedProxies []string `yaml:"trusted_proxies"`
Flows []FlowConfig `yaml:"flows" validate:"min=1,dive,required"`
}
type RateLimiterConfig struct {
Enabled bool `yaml:"enabled"`
Config map[string]interface{} `yaml:"config" validate:"required"`
}
type FlowConfig struct {
Path string `yaml:"path" validate:"required,startswith=/"`
Method string `yaml:"method" validate:"required,oneof=GET POST PUT PATCH DELETE HEAD OPTIONS QUERY"`
Streaming bool `yaml:"streaming"`
// Aggregation is required only for flows with more than one upstream -
// a single-upstream flow is proxied directly and never aggregates
// (enforced in validateFlows, since that depends on len(Upstreams)).
Aggregation *AggregationConfig `yaml:"aggregation"`
Upstreams []UpstreamConfig `yaml:"upstreams" validate:"required,min=1,dive,required"`
Plugins []PluginConfig `yaml:"plugins" validate:"omitempty,dive"`
Middlewares []MiddlewareConfig `yaml:"middlewares" validate:"omitempty,dive"`
}
type AggregationConfig struct {
BestEffort bool `yaml:"best_effort"`
Strategy string `yaml:"strategy" validate:"required,oneof=array merge namespace"`
OnConflict *OnConflictConfig `yaml:"on_conflict" validate:"required_if=Strategy merge"`
}
type OnConflictConfig struct {
Policy string `yaml:"policy" validate:"oneof=overwrite error first prefer"`
Upstream string `yaml:"prefer_upstream" validate:"required_if=Policy prefer"`
}
type UpstreamConfig struct {
Name string `yaml:"name" validate:"required"`
Hosts AddrList `yaml:"hosts" validate:"min=1,dive"`
Path string `yaml:"path"`
Method string `yaml:"method"`
Timeout time.Duration `yaml:"timeout" default:"3s"`
ForwardHeaders []string `yaml:"forward_headers"`
ForwardQueries []string `yaml:"forward_queries"`
ForwardParams []string `yaml:"forward_params"`
Policy PolicyConfig `yaml:"policy"`
Transport TransportConfig `yaml:"transport"`
TLS TLSConfig `yaml:"tls"`
}
type TLSConfig struct {
Enabled bool `yaml:"enabled"`
CertFile string `yaml:"cert_file" validate:"required_with=KeyFile"`
KeyFile string `yaml:"key_file" validate:"required_with=CertFile"`
CAFile string `yaml:"ca_file"`
ServerName string `yaml:"server_name"`
InsecureSkipVerify bool `yaml:"insecure_skip_verify"`
MinVersion string `yaml:"min_version" default:"1.2" validate:"omitempty,oneof=1.2 1.3"`
}
type TransportConfig struct {
MaxIdleConns int `yaml:"max_idle_conns" default:"100"`
MaxIdleConnsPerHost int `yaml:"max_idle_conns_per_host" default:"50"`
IdleConnTimeout time.Duration `yaml:"idle_conn_timeout" default:"90s"`
HTTP2 string `yaml:"http2" default:"auto" validate:"omitempty,oneof=auto on off"`
}
type PluginConfig struct {
Name string `yaml:"name" validate:"required"`
Source string `yaml:"source" validate:"required,oneof=builtin file"`
Path string `yaml:"path" validate:"required_if=Source file"`
Config map[string]interface{} `yaml:"config"`
}
type MiddlewareConfig struct {
Name string `yaml:"name" validate:"required"`
Source string `yaml:"source" validate:"required,oneof=builtin file"`
Path string `yaml:"path" validate:"required_if=Source file,omitempty"`
Config map[string]interface{} `yaml:"config"`
}
type PolicyConfig struct {
HeaderBlacklist []string `yaml:"header_blacklist"`
RequireBody bool `yaml:"require_body"`
MaxResponseBodySize int64 `yaml:"max_response_body_size"`
FollowRedirects bool `yaml:"follow_redirects"`
RetryConfig RetryConfig `yaml:"retry"`
CircuitBreakerConfig CircuitBreakerConfig `yaml:"circuit_breaker"`
LoadBalancingConfig LoadBalancingConfig `yaml:"load_balancing"`
}
type RetryConfig struct {
MaxRetries int `yaml:"max_retries"`
RetryOnStatuses []int `yaml:"retry_on_statuses"`
BackoffDelay time.Duration `yaml:"backoff_delay"`
}
type CircuitBreakerConfig struct {
Enabled bool `yaml:"enabled"`
MaxFailures int `yaml:"max_failures"`
ResetTimeout time.Duration `yaml:"reset_timeout"`
}
type LoadBalancingConfig struct {
Mode string `yaml:"mode"`
}
type AddrList []string
func (a *AddrList) UnmarshalYAML(value *yaml.Node) error {
switch value.Kind {
case yaml.ScalarNode:
var addr string
if err := value.Decode(&addr); err != nil {
return err
}
*a = []string{addr}
return nil
case yaml.SequenceNode:
var addrs []string
if err := value.Decode(&addrs); err != nil {
return err
}
*a = addrs
return nil
case yaml.DocumentNode, yaml.MappingNode, yaml.AliasNode:
return fmt.Errorf("expects a string or sequence, got %v", value.Kind)
default:
return fmt.Errorf("unexpected yaml node kind for AddrList: %v", value.Kind)
}
}
// LoadConfig reads, parses, applies defaults, validates, and returns the config.
func LoadConfig(path string) (Config, error) {
data, err := os.ReadFile(path)
if err != nil {
return Config{}, fmt.Errorf("cannot read configuration file: %w", err)
}
var cfg Config
if err = yaml.Unmarshal(data, &cfg); err != nil {
return Config{}, fmt.Errorf("cannot parse configuration file: %w", err)
}
if err = ValidateConfig(&cfg); err != nil {
return Config{}, err
}
return cfg, nil
}
func DefaultUpstreamConfig() UpstreamConfig {
var u UpstreamConfig
_ = defaults.Set(&u)
return u
}
func ValidateConfig(cfg *Config) error {
if err := defaults.Set(cfg); err != nil {
return fmt.Errorf("cannot apply configuration defaults: %w", err)
}
v := newValidator()
if err := v.Struct(cfg); err != nil {
return fmt.Errorf("invalid configuration:\n%w", formatValidationError(err))
}
if cfg.Gateway.Server.Port == cfg.Gateway.Admin.Port {
return errors.New("server.port and admin.port must differ")
}
if err := validateFlows(*cfg); err != nil {
return fmt.Errorf("invalid flow configuration: %w", err)
}
if err := validateHTTP2(*cfg); err != nil {
return err
}
return nil
}
func newValidator() *validator.Validate {
v := validator.New()
v.RegisterTagNameFunc(func(fld reflect.StructField) string {
name := fld.Tag.Get("yaml")
if name == "" || name == "-" {
return strings.ToLower(fld.Name)
}
return strings.ToLower(strings.Split(name, ",")[0])
})
return v
}
var pathParamPattern = regexp.MustCompile(`\{([^}]+)\}`)
func validateFlows(cfg Config) error {
for _, f := range cfg.Gateway.Routing.Flows {
flowParams := extractPathParams(f.Path)
for _, u := range f.Upstreams {
if err := validateUpstreamParams(u, flowParams, f.Path); err != nil {
return err
}
}
if f.Streaming && len(f.Upstreams) > 1 {
return fmt.Errorf("flow %q: streaming flow must have only one upstream", f.Path)
}
if !f.Streaming && len(f.Upstreams) > 1 && f.Aggregation == nil {
return fmt.Errorf("flow %q: aggregation is required for flows with more than one upstream", f.Path)
}
}
return nil
}
func validateHTTP2(cfg Config) error {
if cfg.Gateway.Server.HTTP2 == "on" && !cfg.Gateway.Server.TLS.Enabled {
return errors.New("gateway.server.http2: cannot be 'on' without gateway.server.tls.enabled")
}
for _, f := range cfg.Gateway.Routing.Flows {
for _, u := range f.Upstreams {
if u.Transport.HTTP2 == "on" && !u.TLS.Enabled {
return fmt.Errorf("upstream %q: transport.http2 cannot be 'on' without tls.enabled", u.Name)
}
}
}
return nil
}
func extractPathParams(path string) map[string]struct{} {
params := make(map[string]struct{})
for _, match := range pathParamPattern.FindAllStringSubmatch(path, -1) {
params[match[1]] = struct{}{}
}
return params
}
func validateUpstreamParams(u UpstreamConfig, flowParams map[string]struct{}, flowPath string) error {
for _, match := range pathParamPattern.FindAllStringSubmatch(u.Path, -1) {
if _, ok := flowParams[match[1]]; !ok {
return fmt.Errorf(
"upstream %q: path param '{%s}' not declared in flow path %q",
u.Name, match[1], flowPath,
)
}
}
for _, param := range u.ForwardParams {
if param == "*" {
continue
}
if _, ok := flowParams[param]; !ok {
return fmt.Errorf(
"upstream %q: forward_param %q not declared in flow path %q",
u.Name, param, flowPath,
)
}
}
return nil
}
func formatValidationError(err error) error {
var ves validator.ValidationErrors
if !errors.As(err, &ves) {
return err
}
messages := make([]string, 0, len(ves))
for _, fe := range ves {
path := strings.TrimPrefix(fe.Namespace(), "Config.")
messages = append(messages, fmt.Sprintf(" %s: %s", path, validationMessage(fe)))
}
return errors.New(strings.Join(messages, "\n"))
}
func validationMessage(fe validator.FieldError) string {
switch fe.Tag() {
case "required":
return "field is required"
case "min":
return fmt.Sprintf("must have at least %s item(s)", fe.Param())
case "oneof":
return fmt.Sprintf("must be one of [%s]", fe.Param())
case "hosts":
return "must be a valid URL"
case "required_if":
if fe.Field() == "path" {
return "is required when source is 'file'"
}
return fe.Error()
default:
return fmt.Sprintf("validation failed on %q", fe.Tag())
}
}