Некоторые проверки не удались
CI / test (push) Successful in 7m31s
Docker / Build and publish worker image (push) Successful in 13m54s
SSH Source-Install E2E / Alpine/Ubuntu/Arch source-install E2E (push) Failing after 30s
342 строки
12 KiB
Go
342 строки
12 KiB
Go
package distworker
|
|
|
|
import (
|
|
"fmt"
|
|
"net/url"
|
|
"os"
|
|
"strconv"
|
|
"strings"
|
|
)
|
|
|
|
// DefaultMaxConcurrency is the default upper bound for the worker pool size
|
|
// and the number of dispatch goroutines that drain the job queue.
|
|
const DefaultMaxConcurrency = 32
|
|
|
|
const (
|
|
// DefaultHTTPHost is the bind address used when WORKER_HOST is unset.
|
|
DefaultHTTPHost = "0.0.0.0"
|
|
// DefaultHTTPPort is the bind port used when WORKER_PORT is unset.
|
|
// Picked >20000 to avoid colliding with the main RSMon app (which
|
|
// binds 7401 by default) when the worker is co-located on the
|
|
// same host. Operators are still free to override via WORKER_PORT.
|
|
DefaultHTTPPort = 27401
|
|
|
|
// EnvPublicURL is the canonical environment variable for the
|
|
// publicly advertised origin. It wins over the legacy
|
|
// EnvWorkerURLLegacy when both are set.
|
|
EnvPublicURL = "PUBLIC_URL"
|
|
// EnvWorkerURLLegacy is the legacy name for the advertised public
|
|
// origin. It is accepted only for the bounded migration defined in
|
|
// docs/public-endpoint-and-identity.md milestone 1 and is
|
|
// deprecated: new configuration must set EnvPublicURL.
|
|
EnvWorkerURLLegacy = "WORKER_URL"
|
|
|
|
// schemeHTTP / schemeHTTPS are the only schemes accepted on
|
|
// PUBLIC_URL. Peer workers and the main app need an http(s) origin
|
|
// they can dial with Go's net/http stack.
|
|
schemeHTTP = "http"
|
|
schemeHTTPS = "https"
|
|
|
|
// loopbackHostnames lists hostnames treated as loopback for the
|
|
// http-on-non-loopback warning emitted by ValidateHTTPConfig.
|
|
loopbackHostnames = "localhost,127.0.0.1,::1,0.0.0.0"
|
|
)
|
|
|
|
// HTTPConfig holds the settings that govern the worker's local HTTP
|
|
// listener (web app MVP in Task 3 and Raft peer connections in Task 4).
|
|
// Host/Port are the bind interface. PublicURL is the publicly-advertised
|
|
// origin peers and the main app use to reach the worker; it is NOT
|
|
// derived from Host:Port because workers commonly sit behind a reverse
|
|
// proxy / Traefik with HTTPS while listening on plain HTTP internally.
|
|
//
|
|
// PublicURL is sourced from PUBLIC_URL (strict origin) or, for the
|
|
// bounded migration, from the deprecated WORKER_URL (legacy-tolerant
|
|
// absolute URL). URLSource records which variable supplied the value so
|
|
// validation and diagnostics name the correct source.
|
|
//
|
|
// Login/Password are basic auth credentials for the worker web app API
|
|
// (Task 3). They are kept in memory only; Task 3 may hash them before
|
|
// any persistent store.
|
|
type HTTPConfig struct {
|
|
Host string
|
|
Port int
|
|
PublicURL string
|
|
URLSource PublicURLSource
|
|
Login string
|
|
Password string
|
|
}
|
|
|
|
// PublicURLSource identifies which environment variable supplied the
|
|
// advertised public origin, so callers can log the legacy deprecation
|
|
// exactly once.
|
|
type PublicURLSource int
|
|
|
|
const (
|
|
// PublicURLSourceNone means no advertised origin is configured.
|
|
PublicURLSourceNone PublicURLSource = iota
|
|
// PublicURLSourceCanonical means PUBLIC_URL was configured.
|
|
PublicURLSourceCanonical
|
|
// PublicURLSourceLegacy means only WORKER_URL was configured; the
|
|
// value is accepted for the bounded migration.
|
|
PublicURLSourceLegacy
|
|
)
|
|
|
|
// PublicURLFromEnv reads the advertised public origin. PUBLIC_URL is
|
|
// the canonical source; the deprecated WORKER_URL is used only when
|
|
// PUBLIC_URL is unset. The returned source lets the caller warn about
|
|
// the legacy fallback without re-reading the environment.
|
|
func PublicURLFromEnv() (string, PublicURLSource) {
|
|
if v := strings.TrimSpace(os.Getenv(EnvPublicURL)); v != "" {
|
|
return v, PublicURLSourceCanonical
|
|
}
|
|
if v := strings.TrimSpace(os.Getenv(EnvWorkerURLLegacy)); v != "" {
|
|
return v, PublicURLSourceLegacy
|
|
}
|
|
return "", PublicURLSourceNone
|
|
}
|
|
|
|
// IsAuthConfigured reports whether both WORKER_LOGIN and WORKER_PASSWORD
|
|
// are set. The HTTP listener refuses to start otherwise.
|
|
func (c HTTPConfig) IsAuthConfigured() bool {
|
|
return c.Login != "" && c.Password != ""
|
|
}
|
|
|
|
// IsListenConfigured reports whether the listener should bind at all.
|
|
// Currently always true while WORKER_PORT > 0; placeholder for Task 3
|
|
// to wire "skip listen" semantics.
|
|
func (c HTTPConfig) IsListenConfigured() bool {
|
|
return c.Port > 0
|
|
}
|
|
|
|
// Config holds the local worker connection settings.
|
|
// Runtime settings are delivered by the control plane over websocket.
|
|
type Config struct {
|
|
URL string
|
|
Token string
|
|
BootstrapToken string
|
|
StateFile string
|
|
WorkerID string
|
|
|
|
// MaxConcurrency caps the worker pool size and the number of dispatcher
|
|
// goroutines. A value <= 0 falls back to DefaultMaxConcurrency.
|
|
// Runtime configuration delivered by the control plane via the websocket
|
|
// "init" or "config" message is clamped to this value when resizing.
|
|
MaxConcurrency int
|
|
|
|
// HTTP holds the local web app / Raft listener settings (Task 2).
|
|
HTTP HTTPConfig
|
|
}
|
|
|
|
// ConfigFromEnv creates a Config from environment variables.
|
|
func ConfigFromEnv() Config {
|
|
return Config{
|
|
URL: normalizeURL(os.Getenv("RSMON_URL")),
|
|
Token: os.Getenv("RSMON_TOKEN"),
|
|
BootstrapToken: os.Getenv("RSMON_BOOTSTRAP_TOKEN"),
|
|
StateFile: strings.TrimSpace(os.Getenv("RSMON_STATE_FILE")),
|
|
WorkerID: strings.TrimSpace(os.Getenv("RSMON_WORKER_ID")),
|
|
HTTP: HTTPConfigFromEnv(),
|
|
}
|
|
}
|
|
|
|
// HTTPConfigFromEnv reads HTTP listener settings from the environment.
|
|
// Empty WORKER_HOST defaults to DefaultHTTPHost; empty WORKER_PORT defaults
|
|
// to DefaultHTTPPort. A malformed WORKER_PORT falls back to the default.
|
|
// The public origin is resolved from PUBLIC_URL (canonical) with the
|
|
// deprecated WORKER_URL as the migration fallback; it is parsed loosely
|
|
// here — ValidateHTTPConfig does the real check.
|
|
func HTTPConfigFromEnv() HTTPConfig {
|
|
port := DefaultHTTPPort
|
|
if raw := strings.TrimSpace(os.Getenv("WORKER_PORT")); raw != "" {
|
|
if v, err := strconv.Atoi(raw); err == nil && v > 0 && v <= 65535 {
|
|
port = v
|
|
}
|
|
}
|
|
host := strings.TrimSpace(os.Getenv("WORKER_HOST"))
|
|
if host == "" {
|
|
host = DefaultHTTPHost
|
|
}
|
|
publicURL, source := PublicURLFromEnv()
|
|
return HTTPConfig{
|
|
Host: host,
|
|
Port: port,
|
|
PublicURL: publicURL,
|
|
URLSource: source,
|
|
Login: os.Getenv("WORKER_LOGIN"),
|
|
Password: os.Getenv("WORKER_PASSWORD"),
|
|
}
|
|
}
|
|
|
|
// ValidateHTTPConfig enforces the local HTTP-listener invariants:
|
|
//
|
|
// - WORKER_LOGIN and WORKER_PASSWORD must be both set or both empty.
|
|
// A mixed state (XOR) is a config bug and must fail fast so an
|
|
// operator notices immediately. Both empty is allowed while the HTTP
|
|
// listener is not started yet.
|
|
// - When the listener would actually start (WORKER_PORT > 0), both
|
|
// must be set; otherwise we are going to expose an unauthenticated
|
|
// endpoint.
|
|
// - A configured advertised origin must validate. The canonical
|
|
// PUBLIC_URL is held to the strict origin rules of ValidatePublicURL
|
|
// and, in an explicitly production environment
|
|
// (DEPLOY_ENV/RSMON_ENV/GO_ENV = production), plain HTTP on a
|
|
// non-loopback host is rejected. A legacy WORKER_URL (URLSource is
|
|
// PublicURLSourceLegacy) is held only to the tolerant
|
|
// ValidateAdvertisedURL so shapes that previously ran keep running;
|
|
// it never fails production startup, preserving the historical
|
|
// warn-only behavior.
|
|
func ValidateHTTPConfig(c HTTPConfig, willListen bool) error {
|
|
loginSet := c.Login != ""
|
|
passSet := c.Password != ""
|
|
if loginSet != passSet {
|
|
return fmt.Errorf(
|
|
"WORKER_LOGIN and WORKER_PASSWORD must be both set or both empty (got login=%s password=%s)",
|
|
boolStr(loginSet), boolStr(passSet))
|
|
}
|
|
if willListen && !c.IsAuthConfigured() {
|
|
return fmt.Errorf("HTTP listener refused to start: WORKER_LOGIN and WORKER_PASSWORD must be set when WORKER_PORT > 0")
|
|
}
|
|
if c.PublicURL == "" {
|
|
return nil
|
|
}
|
|
if c.URLSource == PublicURLSourceLegacy {
|
|
if err := ValidateAdvertisedURL(c.PublicURL); err != nil {
|
|
return fmt.Errorf("%s: %w", EnvWorkerURLLegacy, err)
|
|
}
|
|
return nil
|
|
}
|
|
if err := ValidatePublicURL(c.PublicURL); err != nil {
|
|
return err
|
|
}
|
|
if productionEnv() {
|
|
if host, warn := WarnInsecurePublicURL(c.PublicURL); warn {
|
|
return fmt.Errorf(
|
|
"%s uses plain HTTP on a non-loopback host (%s), which is not permitted in production; front %s with a TLS-terminating reverse proxy",
|
|
EnvPublicURL, host, EnvPublicURL,
|
|
)
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// ValidatePublicURL enforces the strict origin shape required by
|
|
// docs/public-endpoint-and-identity.md ("One Worker, One Public URL")
|
|
// for the canonical PUBLIC_URL configuration: an absolute http(s) URL
|
|
// with a scheme and authority and nothing else. Userinfo, query,
|
|
// fragment, any path other than "/", and a missing hostname (for example
|
|
// "https://:27401") are rejected because every route lives on the origin
|
|
// root and credentials must not leak into the advertised endpoint.
|
|
func ValidatePublicURL(raw string) error {
|
|
u, err := parseURL(raw, EnvPublicURL)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if u.User != nil {
|
|
return fmt.Errorf("%s must not contain userinfo (got %q)", EnvPublicURL, raw)
|
|
}
|
|
if u.RawQuery != "" {
|
|
return fmt.Errorf("%s must not contain a query (got %q)", EnvPublicURL, raw)
|
|
}
|
|
if u.Fragment != "" {
|
|
return fmt.Errorf("%s must not contain a fragment (got %q)", EnvPublicURL, raw)
|
|
}
|
|
if u.Path != "" && u.Path != "/" {
|
|
return fmt.Errorf("%s must not contain a path (got %q)", EnvPublicURL, raw)
|
|
}
|
|
return nil
|
|
}
|
|
|
|
// ValidateAdvertisedURL enforces the tolerant absolute-URL shape used
|
|
// for the legacy WORKER_URL and for endpoints supplied by the control
|
|
// plane: an http(s) URL with a scheme, a host, and a non-empty hostname.
|
|
// Unlike ValidatePublicURL it does not reject a path, userinfo, query,
|
|
// or fragment, because legacy configurations and old control planes may
|
|
// carry such shapes and must not newly fail. It still rejects values
|
|
// that could never be dialed, such as "https://:27401".
|
|
func ValidateAdvertisedURL(raw string) error {
|
|
_, err := parseURL(raw, EnvWorkerURLLegacy)
|
|
return err
|
|
}
|
|
|
|
// parseURL parses an absolute http(s) advertised URL. Both validators
|
|
// share the scheme/host/hostname rules; label names the source in error
|
|
// messages so diagnostics point at the variable the operator set.
|
|
func parseURL(raw, label string) (*url.URL, error) {
|
|
u, err := url.Parse(raw)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("%s is not a valid URL: %v", label, err)
|
|
}
|
|
if u.Scheme == "" || u.Host == "" {
|
|
return nil, fmt.Errorf("%s must be an absolute URL with scheme and host (got %q)", label, raw)
|
|
}
|
|
if u.Scheme != schemeHTTP && u.Scheme != schemeHTTPS {
|
|
return nil, fmt.Errorf("%s scheme must be http or https (got %q)", label, u.Scheme)
|
|
}
|
|
if u.Hostname() == "" {
|
|
return nil, fmt.Errorf("%s must include a host (got %q)", label, raw)
|
|
}
|
|
return u, nil
|
|
}
|
|
|
|
// productionEnv reports whether the worker is explicitly configured for
|
|
// production. Only an explicit literal enables the HTTPS-only PUBLIC_URL
|
|
// policy; an unset variable keeps the historical warn-on-plain-HTTP
|
|
// behavior so existing installations upgrade without surprise.
|
|
func productionEnv() bool {
|
|
for _, key := range []string{"DEPLOY_ENV", "RSMON_ENV", "GO_ENV"} {
|
|
switch strings.ToLower(strings.TrimSpace(os.Getenv(key))) {
|
|
case "production", "prod":
|
|
return true
|
|
}
|
|
}
|
|
return false
|
|
}
|
|
|
|
// WarnInsecurePublicURL logs a warning when PUBLIC_URL uses plain http
|
|
// for a non-loopback host. Returns the host so the caller can log it.
|
|
// A no-op for the loopback case (typical local dev) and for https URLs.
|
|
func WarnInsecurePublicURL(rawURL string) (host string, shouldWarn bool) {
|
|
if rawURL == "" {
|
|
return "", false
|
|
}
|
|
u, err := url.Parse(rawURL)
|
|
if err != nil {
|
|
return "", false
|
|
}
|
|
if u.Scheme != schemeHTTP {
|
|
return u.Host, false
|
|
}
|
|
if isLoopbackHost(u.Hostname()) {
|
|
return u.Host, false
|
|
}
|
|
return u.Host, true
|
|
}
|
|
|
|
func isLoopbackHost(host string) bool {
|
|
host = strings.ToLower(host)
|
|
for _, h := range strings.Split(loopbackHostnames, ",") {
|
|
if host == strings.TrimSpace(h) {
|
|
return true
|
|
}
|
|
}
|
|
return false
|
|
}
|
|
|
|
func boolStr(b bool) string {
|
|
if b {
|
|
return "set"
|
|
}
|
|
return "empty"
|
|
}
|
|
|
|
func normalizeURL(endpoint string) string {
|
|
endpoint = strings.TrimRight(endpoint, "/")
|
|
if strings.HasSuffix(endpoint, "/api/worker") {
|
|
return strings.TrimSuffix(endpoint, "/api/worker")
|
|
}
|
|
if strings.HasSuffix(endpoint, "/worker") {
|
|
return strings.TrimSuffix(endpoint, "/worker")
|
|
}
|
|
return endpoint
|
|
}
|