myxce/src/frontend/apiserver/utils/config.go
Rongsong Shen 97f7c800f1
fix(apiserver): validate, default and document job.terminate_interval (bug-1957)
The loader now stores the effective interval (10 s when unset or 0, LSF's
JOB_TERMINATE_INTERVAL default) and refuses a negative one; app_schema.json
describes the setting; src/docs/xce.md documents kill/stop behaviour and
the interval.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
2026-10-10 10:04:47 +00:00

395 lines
12 KiB
Go

package utils
import (
"crypto/rand"
"encoding/hex"
"errors"
"fmt"
"log"
"net"
"os"
"runtime"
"strconv"
"strings"
"github.com/spf13/viper"
"go.uber.org/zap"
)
type BackendConfig struct {
Address string `yaml:"address"`
Port int `yaml:"port"`
From string `yaml:"from"`
Key string `yaml:"key"`
// Peers lists additional controller candidates for HA mode in "host:port"
// format. When non-empty, RegisterToBackend tries each candidate in order
// and connects to whichever accepts the join as primary. Address+Port
// serve as a single-host fallback when Peers is empty.
Peers []string `yaml:"peers"`
}
type AppServerConfig struct {
AppConfig AppConfig `yaml:"app"`
Security Security `yaml:"security"`
RateLimit RateLimitConfig `yaml:"rate_limit"`
Authentication Authentication `yaml:"authentication"`
JobConfig JobConfig `yaml:"job"`
CacheConfig CacheConfig `yaml:"cache"`
Backend BackendConfig `yaml:"backend"`
}
type AppConfig struct {
AppName string `yaml:"name"`
AppMode string `yaml:"mode"`
HTTPPort string `yaml:"http_port"`
AppVersion string `yaml:"version"`
}
// TokenConfig is the configuration for the token
type TokenConfig struct {
Type string `yaml:"type"`
SecretKey string `yaml:"secret_key"`
TokenExpireTime int `yaml:"token_expire_time"`
}
// Authentication is the configuration for the authentication
type Authentication struct {
Token TokenConfig
}
type Security struct {
HTTPS HTTPSConfig `yaml:"https"`
// TrustedProxies lists the reverse proxies (IP or CIDR) whose
// X-Forwarded-For / X-Real-IP headers set the client IP. Empty (the
// default) trusts no proxy: the client IP is the TCP peer address.
TrustedProxies []string `yaml:"trusted_proxies"`
}
type HTTPSConfig struct {
Enabled bool `yaml:"enabled"`
Port string `yaml:"port"`
CertFile string `yaml:"cert_file"`
KeyFile string `yaml:"key_file"`
}
type RateLimitConfig struct {
Global EndpointLimit `yaml:"global"`
Endpoints map[string]EndpointLimit `yaml:"endpoints"`
}
type EndpointLimit struct {
Rate int `yaml:"rate"` // requests per second
Burst int `yaml:"burst"` // requests per minute
Window string `yaml:"window"` // Time window (second, minute)
}
type WhitelistConfig struct {
Whitelist []string `yaml:"whitelist"`
}
type FileUploadConfig struct {
MaxSize string `yaml:"max_size"` // maximum file size in MB
AllowedTypes []string `yaml:"allowed_types"` // allowed file types
DisallowedTypes []string `yaml:"disallowed_types"` // disallowed file types
}
type JobConfig struct {
MaxArraySize int `yaml:"max_array_size"`
// TerminateInterval is the seconds between the signals of a plain kill
// (SIGINT, SIGTERM, SIGKILL) and a plain stop (SIGTSTP, SIGSTOP), as
// LSF's JOB_TERMINATE_INTERVAL (bug-1957)
TerminateInterval int `yaml:"terminate_interval"`
}
type CORSConfig struct {
Enabled bool `yaml:"enabled"`
AllowedOrigins []string `yaml:"allowed_origins"`
}
type CacheConfig struct {
RefreshInterval int `yaml:"refresh_interval"`
}
var DirPerm os.FileMode
// Add global variables to store configs
var (
AppYamlConfig *AppConfig
AuthenticationYamlConfig *Authentication
HttpsYamlConfig *HTTPSConfig
TrustedProxiesYamlConfig []string
RateLimitYamlConfig *RateLimitConfig
WhitelistYamlConfig *WhitelistConfig
FileUploadYamlConfig *FileUploadConfig
JobYamlConfig *JobConfig
CacheYamlConfig *CacheConfig
CORSYamlConfig *CORSConfig
BackendYamlConfig *BackendConfig
)
func init() {
if runtime.GOOS == "windows" {
DirPerm = 0666
} else {
DirPerm = 0700
}
}
func InitConf() error {
Debug("Enter InitConf() function")
viper.SetConfigType("yaml")
viper.AddConfigPath(Global_TopDir + "/config")
viper.SetConfigName("app")
// validate the app.yaml file whether is validate
data, err := ValidateYamlFormat(Global_TopDir + "/config/app.yaml")
if err != nil {
Error("Fail to call DiagnoseLogCreationError with error")
return err
}
if data == nil {
Error("Fail to call ValidateYamlFormat and data is nil")
return err
}
err1 := viper.ReadInConfig()
if err1 != nil {
_, ok := err1.(viper.ConfigFileNotFoundError)
if ok {
Error("Can't find the configuration files of app.yaml")
return err1
} else {
log.Println("The configuration file of app.yaml with some error")
return err1
}
}
// parse app section
AppYamlConfig = &AppConfig{
AppName: viper.GetString("app.name"),
AppMode: viper.GetString("app.runmode"),
HTTPPort: viper.GetString("app.http_port"),
AppVersion: viper.GetString("app.version"),
}
// parse authentication section
configuredSecret := viper.GetString("authentication.token.secretKey")
jwtSecret := resolveJWTSecret(configuredSecret)
if jwtSecret != configuredSecret && jwtSecret != os.Getenv("XCE_JWT_SECRET") {
Warn("No JWT secret configured (XCE_JWT_SECRET or authentication.token.secretKey), " +
"or it is the publicly known default: using a random per-process secret. " +
"Tokens do not survive an apiserver restart, and apiservers sharing one endpoint " +
"must all set the same XCE_JWT_SECRET")
}
AuthenticationYamlConfig = &Authentication{
Token: TokenConfig{
Type: viper.GetString("authentication.token.type"),
SecretKey: jwtSecret,
TokenExpireTime: int(viper.GetFloat64("authentication.token.token_expire_time")),
},
}
// parse security section
HttpsYamlConfig = &HTTPSConfig{
Enabled: viper.GetBool("security.https.enabled"),
Port: viper.GetString("security.https.port"),
CertFile: viper.GetString("security.https.cert_file"),
KeyFile: viper.GetString("security.https.key_file"),
}
trustedProxies, err := resolveTrustedProxies(viper.GetStringSlice("security.trusted_proxies"))
if err != nil {
return err
}
TrustedProxiesYamlConfig = trustedProxies
// parse rate limit section
RateLimitYamlConfig = &RateLimitConfig{
Global: EndpointLimit{
Rate: viper.GetInt("rate_limit.global.rate"),
Burst: viper.GetInt("rate_limit.global.burst"),
Window: viper.GetString("rate_limit.global.window"),
},
Endpoints: make(map[string]EndpointLimit),
}
// parse endpoints
endpoints := viper.GetStringMap("rate_limit.endpoints")
for path := range endpoints {
RateLimitYamlConfig.Endpoints[path] = EndpointLimit{
Rate: viper.GetInt("rate_limit.endpoints." + path + ".rate"),
Burst: viper.GetInt("rate_limit.endpoints." + path + ".burst"),
Window: viper.GetString("rate_limit.endpoints." + path + ".window"),
}
}
WhitelistYamlConfig = &WhitelistConfig{
Whitelist: viper.GetStringSlice("whitelist"),
}
// parse file upload section
rawMaxSize := viper.GetString("file_upload.max_size")
fileMaxSizeMB, err := ParseMaxSize(rawMaxSize)
if err != nil {
return fmt.Errorf("failed to parse max_size: %v", err)
}
FileUploadYamlConfig = &FileUploadConfig{
MaxSize: strconv.FormatInt(fileMaxSizeMB, 10) + " MB",
AllowedTypes: viper.GetStringSlice("file_upload.allowed_types"),
DisallowedTypes: viper.GetStringSlice("file_upload.disallowed_types"),
}
// parse job config section; an absent section or key reads as 0 (default)
configuredArraySize := viper.GetInt("job.max_array_size")
maxArraySize, clamped, err := resolveMaxArraySize(configuredArraySize)
if err != nil {
return err
}
if clamped {
Warn("job.max_array_size above supported maximum",
zap.Int("configured", configuredArraySize),
zap.Int("effective", maxArraySize))
}
terminateInterval := viper.GetInt("job.terminate_interval")
if terminateInterval < 0 {
return fmt.Errorf("job.terminate_interval must not be negative: %d", terminateInterval)
}
if terminateInterval == 0 {
terminateInterval = defaultTerminateInterval
}
JobYamlConfig = &JobConfig{
MaxArraySize: maxArraySize,
TerminateInterval: terminateInterval,
}
// parse cache config section
if !viper.IsSet("cache") {
// if cache section is not set, use default values
CacheYamlConfig = &CacheConfig{
RefreshInterval: 300,
}
} else {
refreshInterval := viper.GetInt("cache.refresh_interval")
if refreshInterval == 0 {
refreshInterval = 300
}
if refreshInterval < 1 || refreshInterval > 300 {
return fmt.Errorf("invalid cache.refresh_interval value: %d, must be between 1 and 300", refreshInterval)
}
CacheYamlConfig = &CacheConfig{
RefreshInterval: refreshInterval,
}
}
CORSYamlConfig = &CORSConfig{
Enabled: viper.GetBool("cors.enabled"),
AllowedOrigins: viper.GetStringSlice("cors.allowed_origins"),
}
BackendYamlConfig = &BackendConfig{
Address: viper.GetString("backend.address"),
Port: viper.GetInt("backend.port"),
From: viper.GetString("backend.from"),
Key: viper.GetString("backend.key"),
Peers: viper.GetStringSlice("backend.peers"),
}
return nil
}
func ParseMaxSize(maxSize string) (int64, error) {
parts := strings.Fields(maxSize)
if len(parts) != 2 {
return 0, errors.New("invalid max_size format, expected '<number> <unit>'")
}
size, err := strconv.ParseInt(parts[0], 10, 64)
if err != nil {
return 0, errors.New("invalid number in max_size")
}
unit := strings.ToUpper(parts[1])
switch unit {
case "MB":
size *= 1 // MB remains as is
case "GB":
size *= 1024 // convert GB to MB
default:
return 0, errors.New("unsupported unit in max_size, use MB or GB")
}
if size > 10240 {
return 0, errors.New("max_size cannot exceed 10GB (10240 MB)")
}
return size, nil
}
const (
// defaultTerminateInterval is used when job.terminate_interval is unset
// or 0: LSF's JOB_TERMINATE_INTERVAL default (bug-1957).
defaultTerminateInterval = 10
// defaultMaxArraySize is used when job.max_array_size is unset or 0.
defaultMaxArraySize = 1000
// supportedMaxArraySize is the largest array the API server creates
// (array element numbers are 1..32767).
supportedMaxArraySize = 32767
)
// resolveMaxArraySize turns the configured job.max_array_size into the
// effective array size limit: 0 (unset) gives the default 1000, 1..32767 is
// used as is, anything above 32767 is clamped to 32767 (clamped = true, the
// caller logs a warning), and a negative value is an error.
func resolveMaxArraySize(v int) (effective int, clamped bool, err error) {
switch {
case v < 0:
return 0, false, fmt.Errorf("invalid job.max_array_size value: %d, must not be negative", v)
case v == 0:
return defaultMaxArraySize, false, nil
case v > supportedMaxArraySize:
return supportedMaxArraySize, true, nil
default:
return v, false, nil
}
}
// resolveTrustedProxies validates security.trusted_proxies: every entry must
// be an IP address or a CIDR. An empty list means no proxy is trusted.
func resolveTrustedProxies(entries []string) ([]string, error) {
proxies := make([]string, 0, len(entries))
for _, e := range entries {
e = strings.TrimSpace(e)
if e == "" {
continue
}
if net.ParseIP(e) == nil {
if _, _, err := net.ParseCIDR(e); err != nil {
return nil, fmt.Errorf("invalid security.trusted_proxies entry %q: must be an IP address or CIDR", e)
}
}
proxies = append(proxies, e)
}
return proxies, nil
}
// knownDefaultJWTSecret is the secret older app.yaml files shipped with; it is
// public, so it is treated as "no secret configured" (bug-1945).
const knownDefaultJWTSecret = "xce-api-key"
// usableJWTSecret reports whether s may sign tokens: non-empty and not the
// publicly known default.
func usableJWTSecret(s string) bool {
return s != "" && s != knownDefaultJWTSecret
}
// resolveJWTSecret returns the JWT secret: XCE_JWT_SECRET if usable, else
// the config file value if usable, else a random per-process secret (the
// caller detects that case because the result differs from both inputs).
func resolveJWTSecret(configValue string) string {
if env := os.Getenv("XCE_JWT_SECRET"); usableJWTSecret(env) {
return env
}
if usableJWTSecret(configValue) {
return configValue
}
buf := make([]byte, 32)
if _, err := rand.Read(buf); err != nil {
// crypto/rand never fails on supported platforms; refuse to fall
// back to a predictable secret if it ever does.
panic(fmt.Sprintf("cannot generate a JWT secret: %v", err))
}
return hex.EncodeToString(buf)
}