diff --git a/agent/cmd/agent/main.go b/agent/cmd/agent/main.go index f8e12401..59bd84b1 100644 --- a/agent/cmd/agent/main.go +++ b/agent/cmd/agent/main.go @@ -25,6 +25,7 @@ import ( "techulus/cloud-agent/internal/network" "techulus/cloud-agent/internal/paths" "techulus/cloud-agent/internal/reconcile" + "techulus/cloud-agent/internal/routeowners" "techulus/cloud-agent/internal/traefik" "techulus/cloud-agent/internal/wireguard" ) @@ -289,13 +290,14 @@ func main() { var logsSender *logs.VictoriaLogsSender var agentLogWriter *logs.AgentLogWriter var metricsSender agent.MetricsSender + routeOwners := routeowners.NewRegistry() if logsEndpoint != "" { log.Println("[logs] log collection enabled, endpoint:", sanitizedEndpoint(logsEndpoint)) logsSender = logs.NewVictoriaLogsSender(logsEndpoint, config.ServerID) logCollector = logs.NewCollector(logsSender, dataDir) if isProxy { - traefikLogCollector = logs.NewTraefikCollector(logsSender) + traefikLogCollector = logs.NewTraefikCollector(logsSender, routeOwners) log.Println("[traefik-logs] Traefik HTTP log collection enabled") } agentLogWriter = logs.NewAgentLogWriter(config.ServerID, logsSender) @@ -337,7 +339,7 @@ func main() { privateIP := network.PrivateIP() log.Printf("Agent v%s started. Public IP: %s, Private IP: %s. Tick interval: %v", agent.Version, publicIP, privateIP, agent.TickInterval) - agentInstance := agent.NewAgent(client, reconciler, config, publicIP, privateIP, dataDir, logCollector, traefikLogCollector, metricsSender, builder, config.IsProxy, disableDNS) + agentInstance := agent.NewAgent(client, reconciler, config, publicIP, privateIP, dataDir, logCollector, traefikLogCollector, metricsSender, routeOwners, builder, config.IsProxy, disableDNS) agentInstance.Run(ctx) if agentLogFlusherDone != nil { diff --git a/agent/go.mod b/agent/go.mod index 6e7c53d4..bc825dab 100644 --- a/agent/go.mod +++ b/agent/go.mod @@ -9,6 +9,8 @@ require ( github.com/aws/aws-sdk-go-v2/service/s3 v1.71.1 github.com/hashicorp/go-sockaddr v1.0.7 github.com/miekg/dns v1.1.62 + github.com/prometheus/client_model v0.6.2 + github.com/prometheus/common v0.66.1 github.com/shirou/gopsutil/v3 v3.24.5 gopkg.in/yaml.v3 v3.0.1 ) @@ -29,15 +31,19 @@ require ( github.com/aws/aws-sdk-go-v2/service/sts v1.33.3 // indirect github.com/aws/smithy-go v1.22.1 // indirect github.com/go-ole/go-ole v1.2.6 // indirect + github.com/kr/pretty v0.3.1 // indirect github.com/lufia/plan9stats v0.0.0-20211012122336-39d0f177ccd0 // indirect + github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect github.com/power-devops/perfstat v0.0.0-20210106213030-5aafc221ea8c // indirect github.com/shoenig/go-m1cpu v0.1.6 // indirect github.com/tklauser/go-sysconf v0.3.12 // indirect github.com/tklauser/numcpus v0.6.1 // indirect github.com/yusufpapurcu/wmi v1.2.4 // indirect + go.yaml.in/yaml/v2 v2.4.2 // indirect golang.org/x/mod v0.18.0 // indirect - golang.org/x/net v0.27.0 // indirect + golang.org/x/net v0.43.0 // indirect golang.org/x/sync v0.7.0 // indirect - golang.org/x/sys v0.31.0 // indirect + golang.org/x/sys v0.35.0 // indirect golang.org/x/tools v0.22.0 // indirect + google.golang.org/protobuf v1.36.8 // indirect ) diff --git a/agent/go.sum b/agent/go.sum index 5793943c..abcd732c 100644 --- a/agent/go.sum +++ b/agent/go.sum @@ -34,53 +34,73 @@ github.com/aws/aws-sdk-go-v2/service/sts v1.33.3 h1:Xgv/hyNgvLda/M9l9qxXc4UFSgpp github.com/aws/aws-sdk-go-v2/service/sts v1.33.3/go.mod h1:5Gn+d+VaaRgsjewpMvGazt0WfcFO+Md4wLOuBfGR9Bc= github.com/aws/smithy-go v1.22.1 h1:/HPHZQ0g7f4eUeK6HKglFz8uwVfZKgoI25rb/J+dnro= github.com/aws/smithy-go v1.22.1/go.mod h1:irrKGvNn1InZwb2d7fkIRNucdfwR8R+Ts3wxYa/cJHg= +github.com/creack/pty v1.1.9/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ33E= github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= github.com/go-ole/go-ole v1.2.6 h1:/Fpf6oFPoeFik9ty7siob0G6Ke8QvQEuVcuChpwXzpY= github.com/go-ole/go-ole v1.2.6/go.mod h1:pprOEPIfldk/42T2oK7lQ4v4JSDwmV0As9GaiUsvbm0= github.com/google/go-cmp v0.5.6/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE= -github.com/google/go-cmp v0.6.0 h1:ofyhxvXcZhMsU5ulbFiLKl/XBFqE1GSq7atu8tAmTRI= -github.com/google/go-cmp v0.6.0/go.mod h1:17dUlkBOakJ0+DkrSSNjCkIjxS6bF9zb3elmeNGIjoY= +github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8= +github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU= github.com/hashicorp/go-sockaddr v1.0.7 h1:G+pTkSO01HpR5qCxg7lxfsFEZaG+C0VssTy/9dbT+Fw= github.com/hashicorp/go-sockaddr v1.0.7/go.mod h1:FZQbEYa1pxkQ7WLpyXJ6cbjpT8q0YgQaK/JakXqGyWw= +github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE= +github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk= +github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY= +github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE= github.com/lufia/plan9stats v0.0.0-20211012122336-39d0f177ccd0 h1:6E+4a0GO5zZEnZ81pIr0yLvtUWk2if982qA3F3QD6H4= github.com/lufia/plan9stats v0.0.0-20211012122336-39d0f177ccd0/go.mod h1:zJYVVT2jmtg6P3p1VtQj7WsuWi/y4VnjVBn7F8KPB3I= github.com/miekg/dns v1.1.62 h1:cN8OuEF1/x5Rq6Np+h1epln8OiyPWV+lROx9LxcGgIQ= github.com/miekg/dns v1.1.62/go.mod h1:mvDlcItzm+br7MToIKqkglaGhlFMHJ9DTNNWONWXbNQ= +github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 h1:C3w9PqII01/Oq1c1nUAm88MOHcQC9l5mIlSMApZMrHA= +github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822/go.mod h1:+n7T8mK8HuQTcFwEeznm/DIxMOiR9yIdICNftLE1DvQ= +github.com/pkg/diff v0.0.0-20210226163009-20ebb0f2a09e/go.mod h1:pJLUxLENpZxwdsKMEsNbx1VGcRFpLqf3715MtcvvzbA= github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= github.com/power-devops/perfstat v0.0.0-20210106213030-5aafc221ea8c h1:ncq/mPwQF4JjgDlrVEn3C11VoGHZN7m8qihwgMEtzYw= github.com/power-devops/perfstat v0.0.0-20210106213030-5aafc221ea8c/go.mod h1:OmDBASR4679mdNQnz2pUhc2G8CO2JrUAVFDRBDP/hJE= +github.com/prometheus/client_model v0.6.2 h1:oBsgwpGs7iVziMvrGhE53c/GrLUsZdHnqNwqPLxwZyk= +github.com/prometheus/client_model v0.6.2/go.mod h1:y3m2F6Gdpfy6Ut/GBsUqTWZqCUvMVzSfMLjcu6wAwpE= +github.com/prometheus/common v0.66.1 h1:h5E0h5/Y8niHc5DlaLlWLArTQI7tMrsfQjHV+d9ZoGs= +github.com/prometheus/common v0.66.1/go.mod h1:gcaUsgf3KfRSwHY4dIMXLPV0K/Wg1oZ8+SbZk/HH/dA= +github.com/rogpeppe/go-internal v1.9.0/go.mod h1:WtVeX8xhTBvf0smdhujwtBcq4Qrzq/fJaraNFVN+nFs= +github.com/rogpeppe/go-internal v1.10.0 h1:TMyTOH3F/DB16zRVcYyreMH6GnZZrwQVAoYjRBZyWFQ= +github.com/rogpeppe/go-internal v1.10.0/go.mod h1:UQnix2H7Ngw/k4C5ijL5+65zddjncjaFoBhdsK/akog= github.com/shirou/gopsutil/v3 v3.24.5 h1:i0t8kL+kQTvpAYToeuiVk3TgDeKOFioZO3Ztz/iZ9pI= github.com/shirou/gopsutil/v3 v3.24.5/go.mod h1:bsoOS1aStSs9ErQ1WWfxllSeS1K5D+U30r2NfcubMVk= github.com/shoenig/go-m1cpu v0.1.6 h1:nxdKQNcEB6vzgA2E2bvzKIYRuNj7XNJ4S/aRSwKzFtM= github.com/shoenig/go-m1cpu v0.1.6/go.mod h1:1JJMcUBvfNwpq05QDQVAnx3gUHr9IYF7GNg9SUEw2VQ= github.com/shoenig/test v0.6.4 h1:kVTaSd7WLz5WZ2IaoM0RSzRsUD+m8wRR+5qvntpn4LU= github.com/shoenig/test v0.6.4/go.mod h1:byHiCGXqrVaflBLAMq/srcZIHynQPQgeyvkvXnjqq0k= -github.com/stretchr/testify v1.9.0 h1:HtqpIVDClZ4nwg75+f6Lvsy/wHu+3BoSGCbBAcpTsTg= -github.com/stretchr/testify v1.9.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY= +github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U= +github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U= github.com/tklauser/go-sysconf v0.3.12 h1:0QaGUFOdQaIVdPgfITYzaTegZvdCjmYO52cSFAEVmqU= github.com/tklauser/go-sysconf v0.3.12/go.mod h1:Ho14jnntGE1fpdOqQEEaiKRpvIavV0hSfmBq8nJbHYI= github.com/tklauser/numcpus v0.6.1 h1:ng9scYS7az0Bk4OZLvrNXNSAO2Pxr1XXRAPyjhIx+Fk= github.com/tklauser/numcpus v0.6.1/go.mod h1:1XfjsgE2zo8GVw7POkMbHENHzVg3GzmoZ9fESEdAacY= github.com/yusufpapurcu/wmi v1.2.4 h1:zFUKzehAFReQwLys1b/iSMl+JQGSCSjtVqQn9bBrPo0= github.com/yusufpapurcu/wmi v1.2.4/go.mod h1:SBZ9tNy3G9/m5Oi98Zks0QjeHVDvuK0qfxQmPyzfmi0= +go.yaml.in/yaml/v2 v2.4.2 h1:DzmwEr2rDGHl7lsFgAHxmNz/1NlQ7xLIrlN2h5d1eGI= +go.yaml.in/yaml/v2 v2.4.2/go.mod h1:081UH+NErpNdqlCXm3TtEran0rJZGxAYx9hb/ELlsPU= golang.org/x/mod v0.18.0 h1:5+9lSbEzPSdWkH32vYPBwEpX8KwDbM52Ud9xBUvNlb0= golang.org/x/mod v0.18.0/go.mod h1:hTbmBsO62+eylJbnUtE2MGJUyE7QWk4xUqPFrRgJ+7c= -golang.org/x/net v0.27.0 h1:5K3Njcw06/l2y9vpGCSdcxWOYHOUk3dVNGDXN+FvAys= -golang.org/x/net v0.27.0/go.mod h1:dDi0PyhWNoiUOrAS8uXv/vnScO4wnHQO4mj9fn/RytE= +golang.org/x/net v0.43.0 h1:lat02VYK2j4aLzMzecihNvTlJNQUq316m2Mr9rnM6YE= +golang.org/x/net v0.43.0/go.mod h1:vhO1fvI4dGsIjh73sWfUVjj3N7CA9WkKJNQm2svM6Jg= golang.org/x/sync v0.7.0 h1:YsImfSBoP9QPYL0xyKJPq0gcaJdG3rInoqxTWbfQu9M= golang.org/x/sync v0.7.0/go.mod h1:Czt+wKu1gCyEFDUtn0jG5QVvpJ6rzVqr5aXyt9drQfk= golang.org/x/sys v0.0.0-20190916202348-b4ddaad3f8a3/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= golang.org/x/sys v0.0.0-20201204225414-ed752295db88/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= golang.org/x/sys v0.8.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.11.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= -golang.org/x/sys v0.31.0 h1:ioabZlmFYtWhL+TRYpcnNlLwhyxaM9kWTDEmfnprqik= -golang.org/x/sys v0.31.0/go.mod h1:BJP2sWEmIv4KK5OTEluFJCKSidICx8ciO85XgH3Ak8k= +golang.org/x/sys v0.35.0 h1:vz1N37gP5bs89s7He8XuIYXpyY0+QlsKmzipCbUtyxI= +golang.org/x/sys v0.35.0/go.mod h1:BJP2sWEmIv4KK5OTEluFJCKSidICx8ciO85XgH3Ak8k= golang.org/x/tools v0.22.0 h1:gqSGLZqv+AI9lIQzniJ0nZDRG5GBPsSi+DRNHWNz6yA= golang.org/x/tools v0.22.0/go.mod h1:aCwcsjqvq7Yqt6TNyX7QMU2enbQ/Gt0bo6krSeEri+c= golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= -gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM= +google.golang.org/protobuf v1.36.8 h1:xHScyCOEuuwZEc6UtSOvPbAT4zRh0xcNRYekJwfqyMc= +google.golang.org/protobuf v1.36.8/go.mod h1:fuxRtAxBytpl4zzqUh6/eyUujkJdNiuEkXntxiD/uRU= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= +gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk= +gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q= gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= diff --git a/agent/internal/agent/agent.go b/agent/internal/agent/agent.go index 5a56fa73..eeef2be7 100644 --- a/agent/internal/agent/agent.go +++ b/agent/internal/agent/agent.go @@ -11,6 +11,7 @@ import ( agenthttp "techulus/cloud-agent/internal/http" "techulus/cloud-agent/internal/logs" "techulus/cloud-agent/internal/reconcile" + "techulus/cloud-agent/internal/routeowners" ) const ( @@ -44,7 +45,6 @@ type ActualState struct { Containers []container.Container DnsConfigHash string TraefikConfigHash string - L4ConfigHash string CertificatesHash string TraefikReloaded bool ChallengeRouteWritten bool @@ -87,6 +87,7 @@ type Agent struct { LogCollector *logs.Collector TraefikLogCollector *logs.TraefikCollector MetricsSender MetricsSender + RouteOwners *routeowners.Registry Builder *build.Builder isBuilding bool buildMutex sync.Mutex @@ -104,6 +105,7 @@ func NewAgent( logCollector *logs.Collector, traefikLogCollector *logs.TraefikCollector, metricsSender MetricsSender, + routeOwners *routeowners.Registry, builder *build.Builder, isProxy bool, disableDNS bool, @@ -122,6 +124,7 @@ func NewAgent( LogCollector: logCollector, TraefikLogCollector: traefikLogCollector, MetricsSender: metricsSender, + RouteOwners: routeOwners, Builder: builder, IsProxy: isProxy, DisableDNS: disableDNS, diff --git a/agent/internal/agent/drift.go b/agent/internal/agent/drift.go index fe5a3b70..75c02a47 100644 --- a/agent/internal/agent/drift.go +++ b/agent/internal/agent/drift.go @@ -208,7 +208,6 @@ func (a *Agent) getActualState() (*ActualState, error) { } if a.IsProxy { state.TraefikConfigHash = traefik.GetCurrentConfigHash() - state.L4ConfigHash = traefik.GetCurrentL4ConfigHash() state.CertificatesHash = traefik.GetCurrentCertificatesHash() state.TraefikReloaded, err = traefik.DynamicConfigReloaded(a.DataDir) if err != nil { @@ -354,8 +353,8 @@ func (a *Agent) planReconcile(expected *agenthttp.ExpectedState, actual *ActualS if a.IsProxy { compiled := a.compiledTraefikState(expected) - if compiled.HTTPHash != actual.TraefikConfigHash || - compiled.L4Hash != actual.L4ConfigHash || + if compiled.CompileErr != nil || + compiled.RoutesHash != actual.TraefikConfigHash || compiled.CertHash != actual.CertificatesHash || !actual.TraefikReloaded { actions = append(actions, reconcileAction{ @@ -562,6 +561,9 @@ func (a *Agent) applyReconcileAction(action reconcileAction) error { func (a *Agent) updateTraefik() error { compiled := a.compiledTraefikState(a.expectedState) + if compiled.CompileErr != nil { + return fmt.Errorf("failed to compile Traefik routes: %w", compiled.CompileErr) + } needsRestart := false metricsRestart, err := traefik.EnsureMetricsConfig() @@ -585,8 +587,7 @@ func (a *Agent) updateTraefik() error { } } - routesChanged := compiled.HTTPHash != traefik.GetCurrentConfigHash() || - compiled.L4Hash != traefik.GetCurrentL4ConfigHash() + routesChanged := compiled.RoutesHash != traefik.GetCurrentConfigHash() certificatesChanged := compiled.CertHash != traefik.GetCurrentCertificatesHash() if !routesChanged && !certificatesChanged { if err := traefik.EnsureDynamicConfigReloaded(a.DataDir, 15*time.Second); err != nil { @@ -610,7 +611,7 @@ func (a *Agent) updateTraefik() error { } if routesChanged { log.Printf("[reconcile] updating Traefik routes (HTTP: %d, TCP: %d, UDP: %d)", len(compiled.HTTP), len(compiled.TCP), len(compiled.UDP)) - if err := traefik.UpdateHttpRoutesWithL4(compiled.HTTP, compiled.TCP, compiled.UDP, a.expectedState.ServerName); err != nil { + if err := traefik.WriteRoutesConfig(compiled.Routes); err != nil { return fmt.Errorf("failed to update Traefik: %w", err) } } diff --git a/agent/internal/agent/helpers.go b/agent/internal/agent/helpers.go index fecae650..0688c36f 100644 --- a/agent/internal/agent/helpers.go +++ b/agent/internal/agent/helpers.go @@ -59,9 +59,10 @@ type compiledTraefikState struct { TCPPorts []int UDPPorts []int - HTTPHash string - L4Hash string - CertHash string + Routes *traefik.RoutesConfig + RoutesHash string + CompileErr error + CertHash string } func compileTraefikState(expected *agenthttp.ExpectedState) *compiledTraefikState { @@ -86,6 +87,7 @@ func compileTraefikState(expected *agenthttp.ExpectedState) *compiledTraefikStat udpPorts = append(udpPorts, r.ExternalPort) } + routesConfig, compileErr := traefik.CompileRoutes(httpRoutes, tcpRoutes, udpRoutes, expected.ServerName) return &compiledTraefikState{ HTTP: httpRoutes, TCP: tcpRoutes, @@ -93,8 +95,9 @@ func compileTraefikState(expected *agenthttp.ExpectedState) *compiledTraefikStat Certificates: certificates, TCPPorts: tcpPorts, UDPPorts: udpPorts, - HTTPHash: traefik.HashRoutesWithServerName(httpRoutes, expected.ServerName), - L4Hash: traefik.HashTCPRoutes(tcpRoutes) + traefik.HashUDPRoutes(udpRoutes), + Routes: routesConfig, + RoutesHash: traefik.HashRoutesConfig(routesConfig), + CompileErr: compileErr, CertHash: traefik.HashCertificates(certificates), } } diff --git a/agent/internal/agent/reporting.go b/agent/internal/agent/reporting.go index ccd45efa..6dc3fadf 100644 --- a/agent/internal/agent/reporting.go +++ b/agent/internal/agent/reporting.go @@ -164,10 +164,7 @@ func (a *Agent) routingSyncedRolloutIds() []string { func (a *Agent) proxyRoutingStateConverged(expected *agenthttp.ExpectedState) bool { compiled := a.compiledTraefikState(expected) - if compiled.HTTPHash != traefik.GetCurrentConfigHash() { - return false - } - if compiled.L4Hash != traefik.GetCurrentL4ConfigHash() { + if compiled.CompileErr != nil || compiled.RoutesHash != traefik.GetCurrentConfigHash() { return false } if compiled.CertHash != traefik.GetCurrentCertificatesHash() { diff --git a/agent/internal/agent/run.go b/agent/internal/agent/run.go index 522ded73..1f9baa09 100644 --- a/agent/internal/agent/run.go +++ b/agent/internal/agent/run.go @@ -11,7 +11,9 @@ import ( "techulus/cloud-agent/internal/container" agenthttp "techulus/cloud-agent/internal/http" + "techulus/cloud-agent/internal/metrics" "techulus/cloud-agent/internal/serverless" + "techulus/cloud-agent/internal/traefik" ) const ( @@ -23,6 +25,13 @@ const ( ) func (a *Agent) Run(ctx context.Context) { + if a.IsProxy { + if cached, err := a.Client.LoadCachedExpectedState(); err == nil { + a.RouteOwners.Merge(traefik.HTTPRouteOwners(ConvertToHttpRoutes(cached.Traefik.HttpRoutes))) + } else { + log.Printf("[cache] expected state unavailable for initial Traefik attribution: %v", err) + } + } if a.Config.RegistryURL != "" && a.Config.RegistryUsername != "" && a.Config.RegistryPassword != "" { if err := container.Login(a.Config.RegistryURL, a.Config.RegistryUsername, a.Config.RegistryPassword, a.Config.RegistryInsecure); err != nil { log.Printf("[registry] login failed: %v", err) @@ -147,6 +156,10 @@ func (a *Agent) ForwardTraefikMetrics(ctx context.Context) error { return err } + body, err = metrics.EnrichTraefik(body, a.RouteOwners) + if err != nil { + return err + } return a.MetricsSender.SendPrometheusMetrics(body, map[string]string{ "job": "traefik", "server_id": a.Config.ServerID, diff --git a/agent/internal/agent/serverless.go b/agent/internal/agent/serverless.go index 374d6b86..7afa2ea1 100644 --- a/agent/internal/agent/serverless.go +++ b/agent/internal/agent/serverless.go @@ -11,6 +11,7 @@ import ( "techulus/cloud-agent/internal/container" agenthttp "techulus/cloud-agent/internal/http" + "techulus/cloud-agent/internal/traefik" ) var serverlessTransitionCounter atomic.Uint64 @@ -22,6 +23,9 @@ type serverlessTransitionGuard struct { } func (a *Agent) SetLatestExpectedState(state *agenthttp.ExpectedState) { + if state != nil { + a.RouteOwners.Merge(traefik.HTTPRouteOwners(ConvertToHttpRoutes(state.Traefik.HttpRoutes))) + } a.expectedStateMutex.Lock() defer a.expectedStateMutex.Unlock() a.latestExpectedState = state diff --git a/agent/internal/http/client.go b/agent/internal/http/client.go index fa501923..2dd8e55f 100644 --- a/agent/internal/http/client.go +++ b/agent/internal/http/client.go @@ -204,10 +204,30 @@ func (c *Client) cacheExpectedState(state *ExpectedState) error { return err } path := filepath.Join(c.dataDir, expectedStateCacheFile) - return os.WriteFile(path, data, 0600) + if err := os.MkdirAll(c.dataDir, 0700); err != nil { + return err + } + tmp, err := os.CreateTemp(c.dataDir, expectedStateCacheFile+".tmp-") + if err != nil { + return err + } + tmpPath := tmp.Name() + defer os.Remove(tmpPath) + if err := tmp.Chmod(0600); err != nil { + tmp.Close() + return err + } + if _, err := tmp.Write(data); err != nil { + tmp.Close() + return err + } + if err := tmp.Close(); err != nil { + return err + } + return os.Rename(tmpPath, path) } -func (c *Client) loadCachedExpectedState() (*ExpectedState, error) { +func (c *Client) LoadCachedExpectedState() (*ExpectedState, error) { if c.dataDir == "" { return nil, fmt.Errorf("data dir not configured") } @@ -261,7 +281,7 @@ func (c *Client) GetExpectedStateWithFallback() (*ExpectedState, bool, error) { } log.Printf("[state] CP unreachable, attempting to use cached state: %v", err) - cachedState, cacheErr := c.loadCachedExpectedState() + cachedState, cacheErr := c.LoadCachedExpectedState() if cacheErr != nil { return nil, false, fmt.Errorf("CP unreachable and no cached state available: %w (cache error: %v)", err, cacheErr) } diff --git a/agent/internal/logs/traefik_collector.go b/agent/internal/logs/traefik_collector.go index 182f4622..ac0ebccd 100644 --- a/agent/internal/logs/traefik_collector.go +++ b/agent/internal/logs/traefik_collector.go @@ -7,9 +7,10 @@ import ( "io" "log" "os" - "strings" "sync" "time" + + "techulus/cloud-agent/internal/routeowners" ) const ( @@ -59,15 +60,18 @@ type TraefikCollector struct { lastPos int64 initialized bool droppedCount int + unknownCount int + owners *routeowners.Registry } -func NewTraefikCollector(sender HTTPLogSender) *TraefikCollector { +func NewTraefikCollector(sender HTTPLogSender, owners *routeowners.Registry) *TraefikCollector { ctx, cancel := context.WithCancel(context.Background()) return &TraefikCollector{ sender: sender, queue: make([]HTTPLogEntry, 0, traefikMaxBatchSize), ctx: ctx, cancel: cancel, + owners: owners, } } @@ -182,8 +186,12 @@ func (c *TraefikCollector) processLine(line []byte) { return } - serviceId := extractServiceId(entry.RouterName) - if serviceId == "" { + serviceId, ok := c.owners.Lookup(entry.RouterName) + if !ok { + c.unknownCount++ + if c.unknownCount%100 == 1 { + log.Printf("[traefik-logs] dropping log for unknown router (total dropped: %d)", c.unknownCount) + } return } @@ -209,14 +217,6 @@ func (c *TraefikCollector) processLine(line []byte) { c.enqueue(httpEntry) } -func extractServiceId(routerName string) string { - parts := strings.Split(routerName, "@") - if len(parts) > 0 && parts[0] != "" { - return parts[0] - } - return "" -} - func (c *TraefikCollector) enqueue(entry HTTPLogEntry) { c.queueMu.Lock() defer c.queueMu.Unlock() diff --git a/agent/internal/logs/traefik_collector_test.go b/agent/internal/logs/traefik_collector_test.go index 45f7ee1a..40b0ada8 100644 --- a/agent/internal/logs/traefik_collector_test.go +++ b/agent/internal/logs/traefik_collector_test.go @@ -3,10 +3,14 @@ package logs import ( "reflect" "testing" + + "techulus/cloud-agent/internal/routeowners" ) func TestTraefikCollectorProcessLineQueuesRetainedFields(t *testing.T) { - collector := NewTraefikCollector(nil) + owners := routeowners.NewRegistry() + owners.Merge(map[string]string{"http-opaque": "service-42"}) + collector := NewTraefikCollector(nil, owners) collector.processLine([]byte(`{ "ClientHost":"192.0.2.10", "DownstreamContentSize":321, @@ -15,7 +19,7 @@ func TestTraefikCollectorProcessLineQueuesRetainedFields(t *testing.T) { "RequestHost":"api.example.com", "RequestMethod":"POST", "RequestPath":"/v1/items", - "RouterName":"service-42@docker", + "RouterName":"http-opaque@file", "StartUTC":"2026-07-23T10:11:12Z", "time":"", "ClientAddr":"ignored:1234", @@ -39,3 +43,11 @@ func TestTraefikCollectorProcessLineQueuesRetainedFields(t *testing.T) { t.Fatalf("queued entries = %#v, want %#v", collector.queue, want) } } + +func TestTraefikCollectorDropsUnknownRouter(t *testing.T) { + collector := NewTraefikCollector(nil, routeowners.NewRegistry()) + collector.processLine([]byte(`{"RequestHost":"example.com","RouterName":"unknown@file"}`)) + if len(collector.queue) != 0 { + t.Fatal("unknown router was queued") + } +} diff --git a/agent/internal/metrics/traefik.go b/agent/internal/metrics/traefik.go new file mode 100644 index 00000000..c00385b3 --- /dev/null +++ b/agent/internal/metrics/traefik.go @@ -0,0 +1,49 @@ +package metrics + +import ( + "bytes" + "fmt" + + dto "github.com/prometheus/client_model/go" + "github.com/prometheus/common/expfmt" + "github.com/prometheus/common/model" + "techulus/cloud-agent/internal/routeowners" +) + +// EnrichTraefik adds service_id only when Traefik's service label has an +// explicit owner in the registry. Parsing and encoding preserve all metric +// types, escaped labels, and timestamps according to the exposition format. +func EnrichTraefik(data []byte, owners *routeowners.Registry) ([]byte, error) { + parser := expfmt.NewTextParser(model.LegacyValidation) + families, err := parser.TextToMetricFamilies(bytes.NewReader(data)) + if err != nil { + return nil, fmt.Errorf("parse Prometheus metrics: %w", err) + } + var output bytes.Buffer + encoder := expfmt.NewEncoder(&output, expfmt.NewFormat(expfmt.TypeTextPlain)) + for _, family := range families { + for _, metric := range family.Metric { + labels := metric.Label[:0] + var serviceID string + owned := false + for _, label := range metric.Label { + if label.GetName() == "service_id" { + continue + } + if label.GetName() == "service" { + serviceID, owned = owners.Lookup(label.GetValue()) + } + labels = append(labels, label) + } + metric.Label = labels + if owned { + name, value := "service_id", serviceID + metric.Label = append(metric.Label, &dto.LabelPair{Name: &name, Value: &value}) + } + } + if err := encoder.Encode(family); err != nil { + return nil, err + } + } + return output.Bytes(), nil +} diff --git a/agent/internal/metrics/traefik_test.go b/agent/internal/metrics/traefik_test.go new file mode 100644 index 00000000..fba37751 --- /dev/null +++ b/agent/internal/metrics/traefik_test.go @@ -0,0 +1,39 @@ +package metrics + +import ( + "strings" + "testing" + + "techulus/cloud-agent/internal/routeowners" +) + +func TestEnrichTraefik(t *testing.T) { + owners := routeowners.NewRegistry() + owners.Merge(map[string]string{"http-opaque": "service-42"}) + input := `# TYPE requests_total counter +requests_total{service="http-opaque@file",service_id="forged",path="a\\\"b"} 2 123 +requests_total{service="unknown",service_id="forged"} 3 +requests_total{service_id="forged"} 4 +# TYPE latency histogram +latency_bucket{service="http-opaque",le="1"} 1 +latency_sum{service="http-opaque"} 0.5 +latency_count{service="http-opaque"} 1 +` + got, err := EnrichTraefik([]byte(input), owners) + if err != nil { + t.Fatal(err) + } + text := string(got) + if strings.Count(text, `service_id="service-42"`) != 5 { + t.Fatalf("unexpected enrichment:\n%s", text) + } + if strings.Contains(text, `service_id="forged"`) { + t.Fatalf("untrusted service ID was forwarded:\n%s", text) + } + if !strings.Contains(text, `requests_total{service="unknown"} 3`) { + t.Fatalf("unknown service enriched:\n%s", text) + } + if !strings.Contains(text, `requests_total 4`) { + t.Fatalf("metric without a service label was altered incorrectly:\n%s", text) + } +} diff --git a/agent/internal/routeowners/registry.go b/agent/internal/routeowners/registry.go new file mode 100644 index 00000000..9f04f9a7 --- /dev/null +++ b/agent/internal/routeowners/registry.go @@ -0,0 +1,34 @@ +package routeowners + +import ( + "strings" + "sync" +) + +// Registry retains resource ownership for the lifetime of the process. +type Registry struct { + mu sync.RWMutex + owners map[string]string +} + +func NewRegistry() *Registry { + return &Registry{owners: make(map[string]string)} +} + +func (r *Registry) Merge(owners map[string]string) { + r.mu.Lock() + defer r.mu.Unlock() + for resource, serviceID := range owners { + if resource != "" && serviceID != "" { + r.owners[resource] = serviceID + } + } +} + +func (r *Registry) Lookup(resource string) (string, bool) { + resource = strings.TrimSuffix(resource, "@file") + r.mu.RLock() + defer r.mu.RUnlock() + serviceID, ok := r.owners[resource] + return serviceID, ok +} diff --git a/agent/internal/routeowners/registry_test.go b/agent/internal/routeowners/registry_test.go new file mode 100644 index 00000000..13dcff4b --- /dev/null +++ b/agent/internal/routeowners/registry_test.go @@ -0,0 +1,25 @@ +package routeowners + +import "testing" + +func TestRegistryRetainsOwnersAndHandlesFileSuffix(t *testing.T) { + registry := NewRegistry() + registry.Merge(map[string]string{"http-old": "service-old"}) + registry.Merge(map[string]string{ + "http-new": "service-new", + }) + + for resource, want := range map[string]string{ + "http-old@file": "service-old", + "http-new": "service-new", + } { + got, ok := registry.Lookup(resource) + if !ok || got != want { + t.Fatalf("Lookup(%q) = %q, %t; want %q, true", resource, got, ok, want) + } + } + + if _, ok := registry.Lookup("http-old@docker"); ok { + t.Fatal("non-file provider suffix unexpectedly matched") + } +} diff --git a/agent/internal/traefik/l4.go b/agent/internal/traefik/l4.go index 443f38ce..5e94d5c8 100644 --- a/agent/internal/traefik/l4.go +++ b/agent/internal/traefik/l4.go @@ -1,14 +1,10 @@ package traefik import ( - "crypto/sha256" - "encoding/hex" "fmt" "log" "os" "path/filepath" - "sort" - "strings" "gopkg.in/yaml.v3" ) @@ -41,332 +37,111 @@ func ValidateL4Routes(tcpRoutes []TraefikTCPRoute, udpRoutes []TraefikUDPRoute) return nil } -func UpdateHttpRoutesWithL4(httpRoutes []TraefikRoute, tcpRoutes []TraefikTCPRoute, udpRoutes []TraefikUDPRoute, serverName string) error { +// CompileRoutes is the single pure compiler for the durable routes.yaml model. +func CompileRoutes(httpRoutes []TraefikRoute, tcpRoutes []TraefikTCPRoute, udpRoutes []TraefikUDPRoute, serverName string) (*RoutesConfig, error) { if err := ValidateL4Routes(tcpRoutes, udpRoutes); err != nil { - return fmt.Errorf("port validation failed: %w", err) + return nil, fmt.Errorf("port validation failed: %w", err) } - config := traefikFullConfigWithMiddlewares{ - HTTP: httpConfigWithMiddlewares{ - Routers: make(map[string]routerWithMiddleware), - Services: make(map[string]service), - Middlewares: make(map[string]middleware), - }, - TCP: tcpConfig{ - Routers: make(map[string]tcpRouter), - Services: make(map[string]tcpService), - }, - UDP: udpConfig{ - Routers: make(map[string]udpRouter), - Services: make(map[string]udpService), - }, + HTTP: httpConfigWithMiddlewares{Routers: map[string]routerWithMiddleware{}, Services: map[string]service{}, Middlewares: map[string]middleware{}}, + TCP: tcpConfig{Routers: map[string]tcpRouter{}, Services: map[string]tcpService{}}, + UDP: udpConfig{Routers: map[string]udpRouter{}, Services: map[string]udpService{}}, } - var middlewareNames []string if serverName != "" { - config.HTTP.Middlewares["forwarded_server"] = middleware{ - Headers: &headersMiddleware{ - CustomRequestHeaders: map[string]string{ - "X-Forwarded-Server": serverName, - }, - }, - } + config.HTTP.Middlewares["forwarded_server"] = middleware{Headers: &headersMiddleware{CustomRequestHeaders: map[string]string{"X-Forwarded-Server": serverName}}} middlewareNames = []string{"forwarded_server@file"} } - for _, route := range httpRoutes { if len(route.Upstreams) == 0 { continue } - - config.HTTP.Routers[route.ServiceId] = routerWithMiddleware{ - Rule: fmt.Sprintf("Host(`%s`)", route.Domain), - EntryPoints: []string{"websecure"}, - Service: route.ServiceId, - TLS: &tlsConfig{}, - Middlewares: middlewareNames, + name := resourceName("http", route.ServiceId, route.ID) + if _, exists := config.HTTP.Routers[name]; exists { + return nil, duplicateResource("HTTP", name) } - + config.HTTP.Routers[name] = routerWithMiddleware{Rule: fmt.Sprintf("Host(`%s`)", route.Domain), EntryPoints: []string{"websecure"}, Service: name, TLS: &tlsConfig{}, Middlewares: middlewareNames} servers := make([]server, len(route.Upstreams)) for i, upstream := range route.Upstreams { - srv := server{URL: fmt.Sprintf("http://%s", upstream.URL)} + servers[i] = server{URL: fmt.Sprintf("http://%s", upstream.URL)} if upstream.Weight > 0 { - srv.Weight = &upstream.Weight + servers[i].Weight = &route.Upstreams[i].Weight } - servers[i] = srv - } - - config.HTTP.Services[route.ServiceId] = service{ - LoadBalancer: loadBalancer{ - Servers: servers, - }, } + config.HTTP.Services[name] = service{LoadBalancer: loadBalancer{Servers: servers}} } - for _, route := range tcpRoutes { if len(route.Upstreams) == 0 { continue } - - routerName := fmt.Sprintf("tcp_%s_%d", route.ServiceId, route.ExternalPort) - entryPoint := fmt.Sprintf("tcp-%d", route.ExternalPort) - - tcpRtr := tcpRouter{ - Rule: "HostSNI(`*`)", - EntryPoints: []string{entryPoint}, - Service: routerName, + name := resourceName("tcp", route.ServiceId, route.ID) + if _, exists := config.TCP.Routers[name]; exists { + return nil, duplicateResource("TCP", name) } - + router := tcpRouter{Rule: "HostSNI(`*`)", EntryPoints: []string{fmt.Sprintf("tcp-%d", route.ExternalPort)}, Service: name} if route.TLSPassthrough { - tcpRtr.TLS = &tcpTLSConfig{Passthrough: true} + router.TLS = &tcpTLSConfig{Passthrough: true} } - - config.TCP.Routers[routerName] = tcpRtr - + config.TCP.Routers[name] = router servers := make([]tcpServer, len(route.Upstreams)) for i, upstream := range route.Upstreams { servers[i] = tcpServer{Address: upstream} } - - config.TCP.Services[routerName] = tcpService{ - LoadBalancer: tcpLoadBalancer{ - Servers: servers, - }, - } + config.TCP.Services[name] = tcpService{LoadBalancer: tcpLoadBalancer{Servers: servers}} } - for _, route := range udpRoutes { if len(route.Upstreams) == 0 { continue } - - routerName := fmt.Sprintf("udp_%s_%d", route.ServiceId, route.ExternalPort) - entryPoint := fmt.Sprintf("udp-%d", route.ExternalPort) - - config.UDP.Routers[routerName] = udpRouter{ - EntryPoints: []string{entryPoint}, - Service: routerName, + name := resourceName("udp", route.ServiceId, route.ID) + if _, exists := config.UDP.Routers[name]; exists { + return nil, duplicateResource("UDP", name) } - + config.UDP.Routers[name] = udpRouter{EntryPoints: []string{fmt.Sprintf("udp-%d", route.ExternalPort)}, Service: name} servers := make([]udpServer, len(route.Upstreams)) for i, upstream := range route.Upstreams { servers[i] = udpServer{Address: upstream} } - - config.UDP.Services[routerName] = udpService{ - LoadBalancer: udpLoadBalancer{ - Servers: servers, - }, - } + config.UDP.Services[name] = udpService{LoadBalancer: udpLoadBalancer{Servers: servers}} } + normalizeFullConfig(&config) + return &RoutesConfig{config: config}, nil +} - log.Printf("[traefik] updating routes: %d HTTP, %d TCP, %d UDP", len(httpRoutes), len(tcpRoutes), len(udpRoutes)) - - data, err := yaml.Marshal(config) +func WriteRoutesConfig(compiled *RoutesConfig) error { + if compiled == nil { + return fmt.Errorf("routes config is nil") + } + data, err := yaml.Marshal(compiled.config) if err != nil { return fmt.Errorf("failed to marshal traefik config: %w", err) } - - if err := os.MkdirAll(traefikDynamicDir, 0755); err != nil { + if err := os.MkdirAll(dynamicConfigDir, 0755); err != nil { return fmt.Errorf("failed to create dynamic config dir: %w", err) } - - routesPath := filepath.Join(traefikDynamicDir, routesFileName) - tmpPath := routesPath + ".tmp" - - if err := os.WriteFile(tmpPath, data, 0644); err != nil { - return fmt.Errorf("failed to write temp config: %w", err) - } - - if err := os.Rename(tmpPath, routesPath); err != nil { - os.Remove(tmpPath) - return fmt.Errorf("failed to rename config file: %w", err) + routesPath := filepath.Join(dynamicConfigDir, routesFileName) + if err := atomicWrite(routesPath, data, 0644); err != nil { + return fmt.Errorf("failed to write routes config: %w", err) } - log.Printf("[traefik] routes updated successfully") return nil } -func HashTCPRoutes(routes []TraefikTCPRoute) string { - sortedRoutes := make([]TraefikTCPRoute, len(routes)) - copy(sortedRoutes, routes) - sort.Slice(sortedRoutes, func(i, j int) bool { - return sortedRoutes[i].ServiceId < sortedRoutes[j].ServiceId - }) - - var sb strings.Builder - for _, r := range sortedRoutes { - sb.WriteString(r.ServiceId) - sb.WriteString(":") - sb.WriteString(fmt.Sprintf("%d", r.ExternalPort)) - sb.WriteString(":") - sb.WriteString(fmt.Sprintf("%t", r.TLSPassthrough)) - sb.WriteString(":") - sortedUpstreams := make([]string, len(r.Upstreams)) - copy(sortedUpstreams, r.Upstreams) - sort.Strings(sortedUpstreams) - for _, u := range sortedUpstreams { - sb.WriteString(u) - sb.WriteString(",") - } - sb.WriteString("|") - } - hash := sha256.Sum256([]byte(sb.String())) - return hex.EncodeToString(hash[:]) -} - -func HashUDPRoutes(routes []TraefikUDPRoute) string { - sortedRoutes := make([]TraefikUDPRoute, len(routes)) - copy(sortedRoutes, routes) - sort.Slice(sortedRoutes, func(i, j int) bool { - return sortedRoutes[i].ServiceId < sortedRoutes[j].ServiceId - }) - - var sb strings.Builder - for _, r := range sortedRoutes { - sb.WriteString(r.ServiceId) - sb.WriteString(":") - sb.WriteString(fmt.Sprintf("%d", r.ExternalPort)) - sb.WriteString(":") - sortedUpstreams := make([]string, len(r.Upstreams)) - copy(sortedUpstreams, r.Upstreams) - sort.Strings(sortedUpstreams) - for _, u := range sortedUpstreams { - sb.WriteString(u) - sb.WriteString(",") - } - sb.WriteString("|") - } - hash := sha256.Sum256([]byte(sb.String())) - return hex.EncodeToString(hash[:]) -} - -func GetCurrentL4ConfigHash() string { - config, err := readCurrentFullConfig() - if err != nil { - log.Printf("[traefik:hash] failed to read config: %v", err) - return "" - } - - var tcpRoutes []TraefikTCPRoute - for routerName, rtr := range config.TCP.Routers { - var externalPort int - var serviceId string - - fmt.Sscanf(routerName, "tcp_%s_%d", &serviceId, &externalPort) - - for _, ep := range rtr.EntryPoints { - fmt.Sscanf(ep, "tcp-%d", &externalPort) - } - - parts := strings.Split(routerName, "_") - if len(parts) >= 2 { - serviceId = parts[1] - } - - var upstreams []string - if svc, exists := config.TCP.Services[routerName]; exists { - for _, s := range svc.LoadBalancer.Servers { - upstreams = append(upstreams, s.Address) - } - } - - tlsPassthrough := false - if rtr.TLS != nil { - tlsPassthrough = rtr.TLS.Passthrough - } - - tcpRoutes = append(tcpRoutes, TraefikTCPRoute{ - ID: routerName, - ServiceId: serviceId, - Upstreams: upstreams, - ExternalPort: externalPort, - TLSPassthrough: tlsPassthrough, - }) - } - - var udpRoutes []TraefikUDPRoute - for routerName, rtr := range config.UDP.Routers { - var externalPort int - var serviceId string - - for _, ep := range rtr.EntryPoints { - fmt.Sscanf(ep, "udp-%d", &externalPort) - } - - parts := strings.Split(routerName, "_") - if len(parts) >= 2 { - serviceId = parts[1] - } - - var upstreams []string - if svc, exists := config.UDP.Services[routerName]; exists { - for _, s := range svc.LoadBalancer.Servers { - upstreams = append(upstreams, s.Address) - } - } - - udpRoutes = append(udpRoutes, TraefikUDPRoute{ - ID: routerName, - ServiceId: serviceId, - Upstreams: upstreams, - ExternalPort: externalPort, - }) - } - - return HashTCPRoutes(tcpRoutes) + HashUDPRoutes(udpRoutes) -} - func readCurrentFullConfig() (*traefikFullConfigWithMiddlewares, error) { - routesPath := filepath.Join(traefikDynamicDir, routesFileName) - data, err := os.ReadFile(routesPath) + path := filepath.Join(dynamicConfigDir, routesFileName) + data, err := os.ReadFile(path) if err != nil { - if os.IsNotExist(err) { - return &traefikFullConfigWithMiddlewares{ - HTTP: httpConfigWithMiddlewares{ - Routers: make(map[string]routerWithMiddleware), - Services: make(map[string]service), - Middlewares: make(map[string]middleware), - }, - TCP: tcpConfig{ - Routers: make(map[string]tcpRouter), - Services: make(map[string]tcpService), - }, - UDP: udpConfig{ - Routers: make(map[string]udpRouter), - Services: make(map[string]udpService), - }, - }, nil + if !os.IsNotExist(err) { + return nil, err } - return nil, err + config := &traefikFullConfigWithMiddlewares{} + normalizeFullConfig(config) + return config, nil } - var config traefikFullConfigWithMiddlewares if err := yaml.Unmarshal(data, &config); err != nil { return nil, err } - - if config.HTTP.Routers == nil { - config.HTTP.Routers = make(map[string]routerWithMiddleware) - } - if config.HTTP.Services == nil { - config.HTTP.Services = make(map[string]service) - } - if config.HTTP.Middlewares == nil { - config.HTTP.Middlewares = make(map[string]middleware) - } - if config.TCP.Routers == nil { - config.TCP.Routers = make(map[string]tcpRouter) - } - if config.TCP.Services == nil { - config.TCP.Services = make(map[string]tcpService) - } - if config.UDP.Routers == nil { - config.UDP.Routers = make(map[string]udpRouter) - } - if config.UDP.Services == nil { - config.UDP.Services = make(map[string]udpService) - } - + normalizeFullConfig(&config) return &config, nil } diff --git a/agent/internal/traefik/routes.go b/agent/internal/traefik/routes.go index acb8c88b..dd87006d 100644 --- a/agent/internal/traefik/routes.go +++ b/agent/internal/traefik/routes.go @@ -3,45 +3,40 @@ package traefik import ( "crypto/sha256" "encoding/hex" + "encoding/json" "fmt" "log" "sort" - "strings" ) -func HashRoutes(routes []TraefikRoute) string { - sortedRoutes := make([]TraefikRoute, len(routes)) - copy(sortedRoutes, routes) - sort.Slice(sortedRoutes, func(i, j int) bool { - return sortedRoutes[i].ServiceId < sortedRoutes[j].ServiceId - }) +type RoutesConfig struct { + config traefikFullConfigWithMiddlewares +} - var sb strings.Builder - for _, r := range sortedRoutes { - sb.WriteString(r.ServiceId) - sb.WriteString(":") - sb.WriteString(r.Domain) - sb.WriteString(":") - sortedUpstreams := make([]Upstream, len(r.Upstreams)) - copy(sortedUpstreams, r.Upstreams) - sort.Slice(sortedUpstreams, func(i, j int) bool { - return sortedUpstreams[i].URL < sortedUpstreams[j].URL - }) - for _, u := range sortedUpstreams { - sb.WriteString(u.URL) - sb.WriteString("@") - sb.WriteString(fmt.Sprintf("%d", u.Weight)) - sb.WriteString(",") +func resourceName(kind, serviceID, routeID string) string { + hash := sha256.Sum256([]byte(serviceID + "\x00" + routeID)) + return kind + "-" + hex.EncodeToString(hash[:]) +} + +func HTTPRouteOwners(routes []TraefikRoute) map[string]string { + owners := make(map[string]string) + for _, route := range routes { + if len(route.Upstreams) != 0 { + owners[resourceName("http", route.ServiceId, route.ID)] = route.ServiceId } - sb.WriteString("|") } - hash := sha256.Sum256([]byte(sb.String())) - return hex.EncodeToString(hash[:]) + return owners } -func HashRoutesWithServerName(routes []TraefikRoute, serverName string) string { - base := HashRoutes(routes) - hash := sha256.Sum256([]byte(base + "|server:" + serverName)) +func HashRoutesConfig(config *RoutesConfig) string { + if config == nil { + return "" + } + data, err := json.Marshal(config.config) + if err != nil { + return "" + } + hash := sha256.Sum256(data) return hex.EncodeToString(hash[:]) } @@ -51,50 +46,91 @@ func GetCurrentConfigHash() string { log.Printf("[traefik:hash] failed to read config: %v", err) return "" } + return HashRoutesConfig(&RoutesConfig{config: *config}) +} - var routes []TraefikRoute - for serviceId, router := range config.HTTP.Routers { - domain := extractDomainFromRule(router.Rule) - - var upstreams []Upstream - if svc, exists := config.HTTP.Services[serviceId]; exists { - for _, server := range svc.LoadBalancer.Servers { - url := strings.TrimPrefix(server.URL, "http://") - weight := 1 - if server.Weight != nil { - weight = *server.Weight - } - upstreams = append(upstreams, Upstream{ - URL: url, - Weight: weight, - }) - } +func normalizeFullConfig(config *traefikFullConfigWithMiddlewares) { + if config.HTTP.Routers == nil { + config.HTTP.Routers = map[string]routerWithMiddleware{} + } + if config.HTTP.Services == nil { + config.HTTP.Services = map[string]service{} + } + if config.HTTP.Middlewares == nil { + config.HTTP.Middlewares = map[string]middleware{} + } + if config.TCP.Routers == nil { + config.TCP.Routers = map[string]tcpRouter{} + } + if config.TCP.Services == nil { + config.TCP.Services = map[string]tcpService{} + } + if config.UDP.Routers == nil { + config.UDP.Routers = map[string]udpRouter{} + } + if config.UDP.Services == nil { + config.UDP.Services = map[string]udpService{} + } + for key, router := range config.HTTP.Routers { + if router.EntryPoints == nil { + router.EntryPoints = []string{} } - - routes = append(routes, TraefikRoute{ - ID: serviceId, - Domain: domain, - Upstreams: upstreams, - ServiceId: serviceId, + if router.Middlewares == nil { + router.Middlewares = []string{} + } + sort.Strings(router.EntryPoints) + sort.Strings(router.Middlewares) + config.HTTP.Routers[key] = router + } + for key, svc := range config.HTTP.Services { + if svc.LoadBalancer.Servers == nil { + svc.LoadBalancer.Servers = []server{} + } + sort.Slice(svc.LoadBalancer.Servers, func(i, j int) bool { + if svc.LoadBalancer.Servers[i].URL != svc.LoadBalancer.Servers[j].URL { + return svc.LoadBalancer.Servers[i].URL < svc.LoadBalancer.Servers[j].URL + } + return weight(svc.LoadBalancer.Servers[i]) < weight(svc.LoadBalancer.Servers[j]) }) + config.HTTP.Services[key] = svc + } + for key, router := range config.TCP.Routers { + if router.EntryPoints == nil { + router.EntryPoints = []string{} + } + sort.Strings(router.EntryPoints) + config.TCP.Routers[key] = router + } + for key, svc := range config.TCP.Services { + if svc.LoadBalancer.Servers == nil { + svc.LoadBalancer.Servers = []tcpServer{} + } + sort.Slice(svc.LoadBalancer.Servers, func(i, j int) bool { return svc.LoadBalancer.Servers[i].Address < svc.LoadBalancer.Servers[j].Address }) + config.TCP.Services[key] = svc + } + for key, router := range config.UDP.Routers { + if router.EntryPoints == nil { + router.EntryPoints = []string{} + } + sort.Strings(router.EntryPoints) + config.UDP.Routers[key] = router + } + for key, svc := range config.UDP.Services { + if svc.LoadBalancer.Servers == nil { + svc.LoadBalancer.Servers = []udpServer{} + } + sort.Slice(svc.LoadBalancer.Servers, func(i, j int) bool { return svc.LoadBalancer.Servers[i].Address < svc.LoadBalancer.Servers[j].Address }) + config.UDP.Services[key] = svc } - - serverName := extractForwardedServerName(config.HTTP.Middlewares) - - return HashRoutesWithServerName(routes, serverName) } -func extractForwardedServerName(middlewares map[string]middleware) string { - if mw, exists := middlewares["forwarded_server"]; exists && mw.Headers != nil { - if value, ok := mw.Headers.CustomRequestHeaders["X-Forwarded-Server"]; ok { - return value - } +func weight(server server) int { + if server.Weight == nil { + return 0 } - return "" + return *server.Weight } -func extractDomainFromRule(rule string) string { - rule = strings.TrimPrefix(rule, "Host(`") - rule = strings.TrimSuffix(rule, "`)") - return rule +func duplicateResource(kind, name string) error { + return fmt.Errorf("duplicate %s route %s", kind, name) } diff --git a/agent/internal/traefik/routes_test.go b/agent/internal/traefik/routes_test.go new file mode 100644 index 00000000..911dc6ea --- /dev/null +++ b/agent/internal/traefik/routes_test.go @@ -0,0 +1,135 @@ +package traefik + +import ( + "strings" + "testing" +) + +func TestHTTPResourceNameIsOpaqueAndOwnerScoped(t *testing.T) { + name := resourceName("http", "service-42", "my--app.example.com") + if !strings.HasPrefix(name, "http-") { + t.Fatalf("resource name %q is missing its type prefix", name) + } + if strings.Contains(name, "service-42") || strings.Contains(name, "my--app.example.com") { + t.Fatalf("resource name %q exposes semantic identity", name) + } + if got := resourceName("http", "service-42", "my--app.example.com"); got != name { + t.Fatalf("resource name is not deterministic: %q != %q", got, name) + } + if got := resourceName("http", "service-43", "my--app.example.com"); got == name { + t.Fatal("same route ID under different owners produced the same resource name") + } +} + +func TestRoutesConfigRoundTripProducesConvergentHash(t *testing.T) { + originalDir := dynamicConfigDir + t.Cleanup(func() { dynamicConfigDir = originalDir }) + dynamicConfigDir = t.TempDir() + + routes := []TraefikRoute{ + { + ID: "app.example.com", + Domain: "app.example.com", + ServiceId: "service-42", + Upstreams: []Upstream{ + {URL: "10.0.0.2:3000", Weight: 2}, + {URL: "10.0.0.1:3000", Weight: 1}, + }, + }, + { + ID: "www.example.com", + Domain: "www.example.com", + ServiceId: "service-42", + Upstreams: []Upstream{{URL: "10.0.0.1:3000", Weight: 5}}, + }, + } + tcpRoutes := []TraefikTCPRoute{{ + ID: "tcp-route", + ServiceId: "service-tcp", + Upstreams: []string{"10.0.0.2:5432", "10.0.0.1:5432"}, + ExternalPort: TCPPortStart, + TLSPassthrough: true, + }} + udpRoutes := []TraefikUDPRoute{{ + ID: "udp-route", + ServiceId: "service-udp", + Upstreams: []string{"10.0.0.2:8125", "10.0.0.1:8125"}, + ExternalPort: UDPPortStart, + }} + + compiled, err := CompileRoutes(routes, tcpRoutes, udpRoutes, "proxy-1") + if err != nil { + t.Fatal(err) + } + if err := WriteRoutesConfig(compiled); err != nil { + t.Fatal(err) + } + config, err := readCurrentFullConfig() + if err != nil { + t.Fatal(err) + } + if got := len(config.HTTP.Routers); got != 2 { + t.Fatalf("generated %d HTTP routers, want 2", got) + } + if got := len(config.HTTP.Services); got != 2 { + t.Fatalf("generated %d HTTP services, want 2", got) + } + if got := len(config.TCP.Routers); got != 1 { + t.Fatalf("generated %d TCP routers, want 1", got) + } + if got := len(config.UDP.Routers); got != 1 { + t.Fatalf("generated %d UDP routers, want 1", got) + } + forwardedServer := config.HTTP.Middlewares["forwarded_server"] + if forwardedServer.Headers == nil || forwardedServer.Headers.CustomRequestHeaders["X-Forwarded-Server"] != "proxy-1" { + t.Fatalf("forwarded server middleware was not preserved: %#v", forwardedServer) + } + tcpRouter := config.TCP.Routers[resourceName("tcp", "service-tcp", "tcp-route")] + if tcpRouter.TLS == nil || !tcpRouter.TLS.Passthrough { + t.Fatalf("TCP TLS passthrough was not preserved: %#v", tcpRouter) + } + for _, route := range routes { + name := resourceName("http", route.ServiceId, route.ID) + router, exists := config.HTTP.Routers[name] + if !exists { + t.Fatalf("router %q was not generated", name) + } + if router.Service != name { + t.Fatalf("router %q targets %q, want %q", name, router.Service, name) + } + } + + if got, want := GetCurrentConfigHash(), HashRoutesConfig(compiled); got != want { + t.Fatalf("current config hash %q, want %q", got, want) + } + reversedHTTP := []TraefikRoute{routes[1], routes[0]} + reversedHTTP[1].Upstreams = []Upstream{routes[0].Upstreams[1], routes[0].Upstreams[0]} + reversedTCP := append([]TraefikTCPRoute(nil), tcpRoutes...) + reversedTCP[0].Upstreams = []string{tcpRoutes[0].Upstreams[1], tcpRoutes[0].Upstreams[0]} + reversedUDP := append([]TraefikUDPRoute(nil), udpRoutes...) + reversedUDP[0].Upstreams = []string{udpRoutes[0].Upstreams[1], udpRoutes[0].Upstreams[0]} + reversed, err := CompileRoutes(reversedHTTP, reversedTCP, reversedUDP, "proxy-1") + if err != nil { + t.Fatal(err) + } + if HashRoutesConfig(reversed) != HashRoutesConfig(compiled) { + t.Fatal("routes config hash depends on route or upstream order") + } +} + +func TestHashRoutesConfigDoesNotMutateInput(t *testing.T) { + config := &RoutesConfig{config: traefikFullConfigWithMiddlewares{ + HTTP: httpConfigWithMiddlewares{Services: map[string]service{ + "service": {LoadBalancer: loadBalancer{Servers: []server{ + {URL: "http://10.0.0.2:3000"}, + {URL: "http://10.0.0.1:3000"}, + }}}, + }}, + }} + + HashRoutesConfig(config) + servers := config.config.HTTP.Services["service"].LoadBalancer.Servers + if servers[0].URL != "http://10.0.0.2:3000" || servers[1].URL != "http://10.0.0.1:3000" { + t.Fatalf("hashing reordered its input: %#v", servers) + } +} diff --git a/cli/internal/cli/app.go b/cli/internal/cli/app.go index 9cfdedd6..9d09e9f2 100644 --- a/cli/internal/cli/app.go +++ b/cli/internal/cli/app.go @@ -594,7 +594,7 @@ func (a *App) deployCommand() *cobra.Command { output.Field(a.Out, "Operation", result.Operation) output.Field(a.Out, "Status", output.Status(result.Status)) if result.RolloutID != nil && *result.RolloutID != "" { - output.Field(a.Out, "Rollout", output.ShortID(*result.RolloutID)) + output.Field(a.Out, "Rollout", *result.RolloutID) } if result.Operation == "build" { output.Field(a.Out, "Next", "build queued; a rollout starts after it succeeds") @@ -1338,7 +1338,7 @@ func (a *App) runAuthLogin(ctx context.Context, host string) error { output.Field(a.Out, "Host", host) key := "created" if exchange.KeyID != "" { - key = output.ShortID(exchange.KeyID) + key = exchange.KeyID } output.Field(a.Out, "Key", key) return nil @@ -1472,7 +1472,7 @@ func printApplyResult(w io.Writer, result applyResponse) { func printStatus(w io.Writer, value manifest.Manifest, status statusResponse) { fmt.Fprintf(w, "%s/%s/%s\n", value.Project.Slug, value.Environment.Name, value.Service.Name) output.Section(w, "Service") - output.Field(w, "ID", output.ShortID(status.Service.ID)) + output.Field(w, "ID", status.Service.ID) if status.Service.Source.Type == "image" { output.Field(w, "Source", status.Service.Source.Image) } else { @@ -1550,8 +1550,6 @@ func printBuilds(w io.Writer, result map[string]any) { continue } switch field.key { - case "id", "commitSha": - value = output.ShortID(value) case "status": value = output.Status(value) case "startedAt", "completedAt", "createdAt": @@ -1602,8 +1600,8 @@ func printConfiguration(w io.Writer, result map[string]any) { printSchedules(w, current["schedules"]) output.Section(w, "Deployment") - printOptionalField(w, "Revision", shortIDValue(result["activeRevisionId"])) - printOptionalField(w, "Deployment", shortIDValue(result["activeDeploymentId"])) + printOptionalField(w, "Revision", result["activeRevisionId"]) + printOptionalField(w, "Deployment", result["activeDeploymentId"]) printOptionalField(w, "Pending", result["hasPendingChanges"]) changes, _ := result["changes"].([]any) if len(changes) > 0 { @@ -1675,13 +1673,6 @@ func printOptionalField(w io.Writer, label string, value any) { } } -func shortIDValue(value any) any { - if id, ok := value.(string); ok && id != "" { - return output.ShortID(id) - } - return nil -} - func printPlacements(w io.Writer, value any) { placements, _ := value.([]any) output.Field(w, "Placements", len(placements)) @@ -1692,7 +1683,7 @@ func printPlacements(w io.Writer, value any) { } server := placement["serverName"] if server == nil { - server = shortIDValue(placement["serverId"]) + server = placement["serverId"] } fmt.Fprintf(w, " * %v: %v replica(s)\n", server, placement["count"]) } @@ -1886,7 +1877,7 @@ func printRevisions(w io.Writer, result map[string]any) { fmt.Fprintln(w) } if id, ok := revision["id"].(string); ok { - output.Field(w, "ID", output.ShortID(id)) + output.Field(w, "ID", id) } if createdAt, ok := revision["createdAt"].(string); ok { output.Field(w, "Created", output.Timestamp(createdAt)) @@ -1897,7 +1888,7 @@ func printRevisions(w io.Writer, result map[string]any) { if rollout, ok := revision["rollout"].(map[string]any); ok { id, _ := rollout["id"].(string) status, _ := rollout["status"].(string) - value := output.ShortID(id) + value := id if status != "" { value += " (" + output.Status(status) + ")" } @@ -1956,8 +1947,6 @@ func printRolloutFields(w io.Writer, rollout map[string]any) { continue } switch field.key { - case "id": - value = output.ShortID(value) case "status", "currentStage": value = output.Status(value) case "createdAt", "completedAt": diff --git a/cli/internal/cli/app_test.go b/cli/internal/cli/app_test.go index 78aafe46..b2201f0e 100644 --- a/cli/internal/cli/app_test.go +++ b/cli/internal/cli/app_test.go @@ -81,7 +81,7 @@ func TestAuthDeviceExchangeCreatesAPIKeyAndWhoamiUsesIt(t *testing.T) { if body["name"] == "" || body["metadata"] == nil { t.Errorf("exchange body = %#v", body) } - w.Write([]byte(`{"apiKey":"new-secret","keyId":"key-id","name":"CLI test"}`)) + w.Write([]byte(`{"apiKey":"new-secret","keyId":"0400075c-69aa-46c2-bccc-fc172b8c6b28","name":"CLI test"}`)) case "/api/v1/me": if r.Header.Get("X-API-Key") != "new-secret" { t.Errorf("X-API-Key = %q", r.Header.Get("X-API-Key")) @@ -99,9 +99,10 @@ func TestAuthDeviceExchangeCreatesAPIKeyAndWhoamiUsesIt(t *testing.T) { t.Fatal(err) } cfg, _ := auth.NewConfigStore("test").ReadConfig() - if polls != 1 || cfg == nil || cfg.APIKey != "new-secret" || cfg.KeyID != "key-id" { + if polls != 1 || cfg == nil || cfg.APIKey != "new-secret" || cfg.KeyID != "0400075c-69aa-46c2-bccc-fc172b8c6b28" { t.Fatalf("polls=%d config=%#v", polls, cfg) } + assertHumanOutput(t, out.String(), "Key", "0400075c-69aa-46c2-bccc-fc172b8c6b28") out.Reset() if err := execute(app, "auth", "whoami"); err != nil { t.Fatal(err) @@ -451,11 +452,11 @@ func TestDeploySourceNeutralAndMismatch(t *testing.T) { return } posts++ - w.Write([]byte(`{"operation":"rollout","status":"queued"}`)) + w.Write([]byte(`{"operation":"rollout","status":"queued","rolloutId":"2c917d90-4bc1-4274-b3bf-34fed009fc12"}`)) })) defer s.Close() writeConfig(t, s.URL) - app, _ := testApp(t, d, s.Client()) + app, out := testApp(t, d, s.Client()) err := execute(app, "deploy") if tc.mismatch { if err == nil || !strings.Contains(err.Error(), "tc apply") || posts != 0 { @@ -463,6 +464,8 @@ func TestDeploySourceNeutralAndMismatch(t *testing.T) { } } else if err != nil || posts != 1 { t.Fatalf("err=%v posts=%d", err, posts) + } else { + assertHumanOutput(t, out.String(), "Rollout", "2c917d90-4bc1-4274-b3bf-34fed009fc12") } }) } @@ -487,7 +490,7 @@ func TestStatusAndResourceRoutesAndOutput(t *testing.T) { var gotPath, gotQuery string s := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { gotPath, gotQuery = r.URL.Path, r.URL.RawQuery - w.Write([]byte(`{"service":{"id":"s","name":"web","source":{"type":"image","image":"nginx"}},"latestBuild":null,"latestRollout":null,"deployments":[],"items":[]}`)) + w.Write([]byte(`{"service":{"id":"0400075c-69aa-46c2-bccc-fc172b8c6b28","name":"web","source":{"type":"image","image":"nginx"}},"latestBuild":null,"latestRollout":null,"deployments":[],"items":[]}`)) })) defer s.Close() writeConfig(t, s.URL) @@ -503,6 +506,14 @@ func TestStatusAndResourceRoutesAndOutput(t *testing.T) { t.Fatalf("invalid JSON: %s", out.String()) } } + if tc.args[0] == "status" { + app, out := testApp(t, t.TempDir(), s.Client()) + args := append(tc.args, "--project", "p", "--environment", "e", "--service", "s") + if err := execute(app, args...); err != nil { + t.Fatal(err) + } + assertHumanOutput(t, out.String(), "ID", "0400075c-69aa-46c2-bccc-fc172b8c6b28") + } base := "/api/v1/projects/p/environments/e/services/s" if gotPath != base+tc.path || gotQuery != tc.query { t.Fatalf("got %s?%s", gotPath, gotQuery) @@ -525,7 +536,7 @@ func TestBuildsHumanOutputIsFormatted(t *testing.T) { { name: "github source", response: `{"supported":true,"builds":[{"id":"0400075c-69aa-46c2-bccc-fc172b8c6b28","status":"in_progress","branch":"main","commitSha":"abcdef1234567890abcdef1234567890abcdef12","createdAt":"2026-01-01T00:00:00.820Z"}],"nextCursor":"next-page"}`, - want: []string{"Builds (1)", "ID", "0400075c...6b28", "Status", "in progress", "Branch", "main", "Commit", "abcdef12...ef12", "Created", "2026-01-01T00:00:00.82Z", "next-page"}, + want: []string{"Builds (1)", "ID", "0400075c-69aa-46c2-bccc-fc172b8c6b28", "Status", "in progress", "Branch", "main", "Commit", "abcdef1234567890abcdef1234567890abcdef12", "Created", "2026-01-01T00:00:00.82Z", "next-page"}, }, } { t.Run(tc.name, func(t *testing.T) { @@ -543,7 +554,7 @@ func TestBuildsHumanOutputIsFormatted(t *testing.T) { } func TestConfigurationHumanOutputIsFormatted(t *testing.T) { - s := responseServer(t, `{"current":{"source":{"type":"github","repository":"https://github.com/acme/app","branch":"main","rootDir":"cmd/api"},"hostname":"api.example.com","stateful":true,"replicas":2,"placements":[{"serverId":"server-1","serverName":"ubuntu-2","count":2}],"healthCheck":{"cmd":"curl localhost:3000","interval":10,"timeout":5,"retries":3,"startPeriod":30},"startCommand":"npm start","resources":{"cpuCores":2,"memoryMb":512},"ports":[{"containerPort":3000,"public":true,"domain":"api.example.com","protocol":"http","externalPort":null}],"volumes":[{"name":"data","containerPath":"/data"}],"serverless":{"enabled":false,"sleepAfterSeconds":300,"wakeTimeoutSeconds":30},"schedules":{"deployment":null,"backup":{"enabled":false,"schedule":null}}},"active":null,"activeRevisionId":"0400075c-69aa-46c2-bccc-fc172b8c6b28","activeDeploymentId":"2c917d90-4bc1-4274-b3bf-34fed009fc12","hasPendingChanges":true,"changes":[{"field":"replicas","from":"active revision","to":"current configuration"}],"management":{"patchable":true,"blockers":[]}}`) + s := responseServer(t, `{"current":{"source":{"type":"github","repository":"https://github.com/acme/app","branch":"main","rootDir":"cmd/api"},"hostname":"api.example.com","stateful":true,"replicas":2,"placements":[{"serverId":"server-1","serverName":"ubuntu-2","count":2},{"serverId":"a868807a-42ee-4a40-99fe-cd8303036b02","count":1}],"healthCheck":{"cmd":"curl localhost:3000","interval":10,"timeout":5,"retries":3,"startPeriod":30},"startCommand":"npm start","resources":{"cpuCores":2,"memoryMb":512},"ports":[{"containerPort":3000,"public":true,"domain":"api.example.com","protocol":"http","externalPort":null}],"volumes":[{"name":"data","containerPath":"/data"}],"serverless":{"enabled":false,"sleepAfterSeconds":300,"wakeTimeoutSeconds":30},"schedules":{"deployment":null,"backup":{"enabled":false,"schedule":null}}},"active":null,"activeRevisionId":"0400075c-69aa-46c2-bccc-fc172b8c6b28","activeDeploymentId":"2c917d90-4bc1-4274-b3bf-34fed009fc12","hasPendingChanges":true,"changes":[{"field":"replicas","from":"active revision","to":"current configuration"}],"management":{"patchable":true,"blockers":[]}}`) writeConfig(t, s.URL) app, out := testApp(t, t.TempDir(), s.Client()) @@ -551,7 +562,7 @@ func TestConfigurationHumanOutputIsFormatted(t *testing.T) { if err != nil { t.Fatal(err) } - assertHumanOutput(t, out.String(), "Configuration", "https://github.com/acme/app @ main (cmd/api)", "api.example.com", "ubuntu-2: 2 replica(s)", "3000/http public", "data: /data", "curl localhost:3000", "Deployment", "0400075c...6b28", "replicas: active revision -> current configuration", "Management", "Patchable", "yes") + assertHumanOutput(t, out.String(), "Configuration", "https://github.com/acme/app @ main (cmd/api)", "api.example.com", "ubuntu-2: 2 replica(s)", "a868807a-42ee-4a40-99fe-cd8303036b02: 1 replica(s)", "3000/http public", "data: /data", "curl localhost:3000", "Deployment", "0400075c-69aa-46c2-bccc-fc172b8c6b28", "2c917d90-4bc1-4274-b3bf-34fed009fc12", "replicas: active revision -> current configuration", "Management", "Patchable", "yes") } func TestMetricsHumanOutputIsFormatted(t *testing.T) { @@ -590,7 +601,7 @@ func TestRevisionsHumanOutputIsFormatted(t *testing.T) { if err != nil { t.Fatal(err) } - assertHumanOutput(t, out.String(), "Revisions (2)", "0400075c...6b28", "Arjun Komath", "in progress", "Image: app:v1 -> app:v2", "3f9293e9...b056", "@octocat", "initial revision", "next-page") + assertHumanOutput(t, out.String(), "Revisions (2)", "0400075c-69aa-46c2-bccc-fc172b8c6b28", "2c917d90-4bc1-4274-b3bf-34fed009fc12", "Arjun Komath", "in progress", "Image: app:v1 -> app:v2", "3f9293e9-d4d8-4b72-8ae4-94b3737ab056", "@octocat", "initial revision", "next-page") } func TestRolloutHumanOutputIsFormatted(t *testing.T) { @@ -613,8 +624,8 @@ func TestRolloutHumanOutputIsFormatted(t *testing.T) { args []string want []string }{ - {[]string{"rollouts"}, []string{"Rollouts (1)", "2c917d90...fc12", "in progress", "health check", "ubuntu-2: running, healthy", "next-page"}}, - {[]string{"rollout", "r1"}, []string{"Rollout", "2c917d90...fc12", "completed", "2026-07-21T11:05:30Z"}}, + {[]string{"rollouts"}, []string{"Rollouts (1)", "2c917d90-4bc1-4274-b3bf-34fed009fc12", "in progress", "health check", "ubuntu-2: running, healthy", "next-page"}}, + {[]string{"rollout", "r1"}, []string{"Rollout", "2c917d90-4bc1-4274-b3bf-34fed009fc12", "completed", "2026-07-21T11:05:30Z"}}, {[]string{"rollout", "logs", "r1"}, []string{"Rollout logs (2)", "[preparing]", "Rollout started", "[health check]", "Starting container"}}, } { app, out := testApp(t, t.TempDir(), s.Client()) diff --git a/cli/internal/manifest/manifest.go b/cli/internal/manifest/manifest.go index 76136a77..d929bb54 100644 --- a/cli/internal/manifest/manifest.go +++ b/cli/internal/manifest/manifest.go @@ -264,15 +264,12 @@ func Validate(m Manifest) error { return errors.New("service.placement.mode must be automatic or manual") } } - seenPorts := make(map[int]struct{}, len(m.Service.Ports)) + portsByNumber := make(map[int][]Port, len(m.Service.Ports)) + seenDomains := make(map[string]struct{}, len(m.Service.Ports)) for i, p := range m.Service.Ports { if p.ContainerPort < 1 || p.ContainerPort > 65535 { return fmt.Errorf("service.ports[%d].containerPort must be between 1 and 65535", i) } - if _, exists := seenPorts[p.ContainerPort]; exists { - return fmt.Errorf("service.ports[%d].containerPort must be unique", i) - } - seenPorts[p.ContainerPort] = struct{}{} if p.Domain != nil && strings.TrimSpace(*p.Domain) == "" { return fmt.Errorf("service.ports[%d].domain cannot be blank", i) } @@ -282,6 +279,24 @@ func Validate(m Manifest) error { if !p.Public && p.Domain != nil { return fmt.Errorf("service.ports[%d].domain cannot be set for internal ports", i) } + if p.Domain != nil { + domain := strings.ToLower(strings.TrimSpace(*p.Domain)) + if _, exists := seenDomains[domain]; exists { + return fmt.Errorf("service.ports[%d].domain must be unique", i) + } + seenDomains[domain] = struct{}{} + } + portsByNumber[p.ContainerPort] = append(portsByNumber[p.ContainerPort], p) + } + for containerPort, ports := range portsByNumber { + if len(ports) < 2 { + continue + } + for _, port := range ports { + if !port.Public || port.Domain == nil { + return fmt.Errorf("service port %d can only be repeated for public domains", containerPort) + } + } } if h := m.Service.HealthCheck; h != nil && (h.Cmd == "" || h.Interval < 1 || h.Timeout < 1 || h.Retries < 1 || h.StartPeriod < 0) { return errors.New("service.healthCheck contains invalid values") diff --git a/cli/internal/manifest/manifest_test.go b/cli/internal/manifest/manifest_test.go index 83e3a025..b18b5709 100644 --- a/cli/internal/manifest/manifest_test.go +++ b/cli/internal/manifest/manifest_test.go @@ -125,7 +125,32 @@ func TestRejectWindowsAbsoluteRootDir(t *testing.T) { func TestRejectDuplicatePorts(t *testing.T) { m := base() m.Service.Ports = []Port{{ContainerPort: 8080}, {ContainerPort: 8080}} - if err := Validate(m); err == nil || !strings.Contains(err.Error(), "must be unique") { + if err := Validate(m); err == nil || !strings.Contains(err.Error(), "can only be repeated") { + t.Fatalf("error = %v", err) + } +} + +func TestAllowsMultipleDomainsOnOnePort(t *testing.T) { + m := base() + first := "app.example.com" + second := "www.example.com" + m.Service.Ports = []Port{ + {ContainerPort: 8080, Public: true, Domain: &first}, + {ContainerPort: 8080, Public: true, Domain: &second}, + } + if err := Validate(m); err != nil { + t.Fatalf("valid port aliases rejected: %v", err) + } +} + +func TestRejectsDuplicatePortDomains(t *testing.T) { + m := base() + domain := "app.example.com" + m.Service.Ports = []Port{ + {ContainerPort: 8080, Public: true, Domain: &domain}, + {ContainerPort: 8080, Public: true, Domain: &domain}, + } + if err := Validate(m); err == nil || !strings.Contains(err.Error(), "domain must be unique") { t.Fatalf("error = %v", err) } } diff --git a/cli/internal/output/output.go b/cli/internal/output/output.go index b961b31d..853e18cc 100644 --- a/cli/internal/output/output.go +++ b/cli/internal/output/output.go @@ -45,13 +45,6 @@ func Next(w io.Writer, command string) { Field(w, "Run", command) } -func ShortID(id string) string { - if len(id) <= 16 { - return id - } - return id[:8] + "..." + id[len(id)-4:] -} - func Status(value string) string { return strings.ReplaceAll(value, "_", " ") } diff --git a/docs/api/public-api.mdx b/docs/api/public-api.mdx index 5fb9c8ae..2f2dd5cc 100644 --- a/docs/api/public-api.mdx +++ b/docs/api/public-api.mdx @@ -209,7 +209,7 @@ Use manual placement to choose exact servers: } ``` -Manual placement requires online servers with WireGuard configured. Serverless services require proxy servers. Automatic placement is not available for stateful, serverless, or volume-backed services. Submit replica changes through `placement`; the API rejects a top-level `replicas` field. +Manual placement requires online servers with WireGuard configured. Serverless services require proxy servers, including when automatic placement is used. Automatic placement is not available for stateful or volume-backed services. Submit replica changes through `placement`; the API rejects a top-level `replicas` field. The API only manages stateless services with HTTP ports. Existing volumes, stateful mode, TCP or UDP ports, TLS passthrough, or invalid resource limits return a conflict with an actionable code. diff --git a/docs/architecture.mdx b/docs/architecture.mdx index c4eea04f..14333107 100644 --- a/docs/architecture.mdx +++ b/docs/architecture.mdx @@ -210,8 +210,9 @@ Worker nodes do not run Traefik. Public HTTP services can be configured as serverless. Serverless scale-to-zero is proxy-local: deployments placed on proxy nodes may sleep after an idle period, -then wake on the next public request handled by that proxy. Serverless services -use manual placement on proxy nodes. +then wake on the next public request handled by that proxy. Stateless services +without volumes can use automatic placement across proxy nodes. Stateful or +volume-backed serverless services require manual placement on a proxy node. Serverless uses the same declarative expected-state model as normal deployments, but proxy agents own the local lifecycle decision: diff --git a/docs/services/scaling.mdx b/docs/services/scaling.mdx index cb1e3cb3..13ab4163 100644 --- a/docs/services/scaling.mdx +++ b/docs/services/scaling.mdx @@ -42,9 +42,9 @@ stores the desired replica count and distributes replicas across online, configured nodes during rollout. Manual placement selects exact target servers and replica counts. -Serverless services currently require manual placement on proxy nodes. Automatic -serverless placement will remain unavailable until every healthy proxy can route -requests through the owning proxy's wake gateway. +Serverless services support automatic placement across online, configured proxy +nodes. Public ingress must use health checks to avoid proxy nodes that do not +currently own a replica for the service. ## Server Pinning diff --git a/docs/services/volumes.mdx b/docs/services/volumes.mdx index d516e02e..583110a2 100644 --- a/docs/services/volumes.mdx +++ b/docs/services/volumes.mdx @@ -18,7 +18,9 @@ Each volume has a name and a container path: When you add a volume, the service automatically becomes **stateful**. Stateful services are locked to a single server and limited to 1 replica so the container always mounts the same local data path. When the last volume is removed, the service reverts to stateless. -Stateful services can use serverless scaling when they have a public HTTP domain. Because volumes are local to one server, only a proxy-hosted stateful replica can sleep and wake on request. A stateful replica placed on a worker node stays always on. +Stateful services can use serverless scaling when they have a public HTTP domain +and manual placement on a proxy node. Automatic serverless placement is only +available to stateless services without volumes. ## Volume Backups diff --git a/web/actions/projects.ts b/web/actions/projects.ts index 3f678905..b95fbb29 100644 --- a/web/actions/projects.ts +++ b/web/actions/projects.ts @@ -51,6 +51,7 @@ import type { HealthCheckConfig as ServiceHealthCheckConfig, } from "@/lib/service-config"; import { MIN_SERVERLESS_SLEEP_AFTER_SECONDS } from "@/lib/service-config"; +import { findServicePortValidationIssue } from "@/lib/service-revision-spec"; import type { DeleteConfirmation } from "@/lib/two-factor"; import { getZodErrorMessage, slugify } from "@/lib/utils"; import { @@ -925,11 +926,6 @@ export async function updateServiceServerlessSettings( } if (validated.enabled) { - if (service.placementMode === "automatic") { - throw new Error( - "Switch to manual placement before enabling serverless", - ); - } const publicHttpPorts = await tx .select({ id: servicePorts.id }) .from(servicePorts) @@ -949,24 +945,43 @@ export async function updateServiceServerlessSettings( ); } - const configuredReplicas = await tx - .select({ - count: serviceReplicas.count, - serverIsProxy: servers.isProxy, - }) - .from(serviceReplicas) - .innerJoin(servers, eq(serviceReplicas.serverId, servers.id)) - .where(eq(serviceReplicas.serviceId, serviceId)); - const totalConfiguredReplicas = configuredReplicas.reduce( - (total, replica) => total + replica.count, - 0, - ); + const [configuredReplicas, volume] = await Promise.all([ + tx + .select({ + count: serviceReplicas.count, + serverIsProxy: servers.isProxy, + }) + .from(serviceReplicas) + .innerJoin(servers, eq(serviceReplicas.serverId, servers.id)) + .where(eq(serviceReplicas.serviceId, serviceId)), + tx + .select({ id: serviceVolumes.id }) + .from(serviceVolumes) + .where(eq(serviceVolumes.serviceId, serviceId)) + .limit(1), + ]); + if ( + service.placementMode === "automatic" && + (service.stateful || volume.length > 0) + ) { + throw new Error( + "Automatic placement is not supported for stateful services or services with volumes", + ); + } + const totalConfiguredReplicas = + service.placementMode === "automatic" + ? service.replicas + : configuredReplicas.reduce( + (total, replica) => total + replica.count, + 0, + ); if (totalConfiguredReplicas < 1) { throw new Error("Serverless services require at least one replica"); } if ( + service.placementMode === "manual" && configuredReplicas.some( (replica) => replica.count > 0 && !replica.serverIsProxy, ) @@ -1083,78 +1098,118 @@ export async function updateServiceConfig( } if (config.ports) { - if (config.ports.remove && config.ports.remove.length > 0) { - for (const portId of config.ports.remove) { - await db.delete(servicePorts).where(eq(servicePorts.id, portId)); - } - } - - if (config.ports.add && config.ports.add.length > 0) { - const existing = await db - .select() - .from(servicePorts) - .where(eq(servicePorts.serviceId, serviceId)); - - for (const port of config.ports.add) { - const protocol = port.protocol || "http"; - + try { + await db.transaction(async (tx) => { + await tx.execute( + sql`SELECT pg_advisory_xact_lock(hashtext(${serviceId}))`, + ); + const existing = await tx + .select() + .from(servicePorts) + .where(eq(servicePorts.serviceId, serviceId)); + const removedIds = new Set(config.ports?.remove ?? []); if ( - existing.some((p) => p.port === port.port && p.protocol === protocol) + [...removedIds].some((id) => !existing.some((port) => port.id === id)) ) { - throw new Error(`Port ${port.port} (${protocol}) already exists`); + throw new Error("Port not found"); } - if (port.isPublic) { - if (protocol === "http") { - if (!port.domain) { - throw new Error("Domain is required for public HTTP ports"); - } - - const domain = port.domain.trim().toLowerCase(); - if (!domain) { - throw new Error("Invalid domain"); - } - - const existingDomain = await db - .select() - .from(servicePorts) - .where(eq(servicePorts.domain, domain)); - - if (existingDomain.length > 0) { - throw new Error("Domain already in use"); - } - - await db.insert(servicePorts).values({ - id: randomUUID(), - serviceId, - port: port.port, - isPublic: true, - domain, - protocol: "http", - }); - } else if (protocol === "tcp" || protocol === "udp") { - const externalPort = await allocatePort(protocol); - - await db.insert(servicePorts).values({ - id: randomUUID(), - serviceId, - port: port.port, - isPublic: true, - protocol, - externalPort, - tlsPassthrough: port.tlsPassthrough ?? false, - }); + const additions: (typeof servicePorts.$inferInsert)[] = []; + const reservedExternalPorts = { + tcp: new Set(), + udp: new Set(), + }; + if ( + (config.ports?.add ?? []).some( + (port) => + port.isPublic && + (port.protocol === "tcp" || port.protocol === "udp"), + ) + ) { + await tx.execute( + sql`SELECT pg_advisory_xact_lock(hashtext('service_port_external_allocation'))`, + ); + } + for (const port of config.ports?.add ?? []) { + const protocol = port.protocol ?? "http"; + const domain = port.domain?.trim().toLowerCase() || null; + const externalPort = + port.isPublic && (protocol === "tcp" || protocol === "udp") + ? await allocatePort( + tx, + protocol, + reservedExternalPorts[protocol], + ) + : null; + if ( + externalPort !== null && + (protocol === "tcp" || protocol === "udp") + ) { + reservedExternalPorts[protocol].add(externalPort); } - } else { - await db.insert(servicePorts).values({ + additions.push({ id: randomUUID(), serviceId, port: port.port, - isPublic: false, + isPublic: port.isPublic, + domain: protocol === "http" && port.isPublic ? domain : null, protocol, + externalPort, + tlsPassthrough: + protocol === "tcp" ? (port.tlsPassthrough ?? false) : false, }); } + + const finalPorts = [ + ...existing.filter((port) => !removedIds.has(port.id)), + ...additions, + ]; + const issue = findServicePortValidationIssue( + finalPorts.map((port) => ({ + containerPort: port.port, + isPublic: port.isPublic ?? false, + domain: port.domain ?? null, + protocol: port.protocol ?? "http", + })), + ); + if (issue) throw new Error(issue.message); + + for (const port of additions) { + if (!port.domain) continue; + const conflict = await tx + .select({ serviceId: servicePorts.serviceId }) + .from(servicePorts) + .where(eq(servicePorts.domain, port.domain)) + .limit(1) + .then((rows) => rows[0]); + if (conflict && conflict.serviceId !== serviceId) { + throw new Error("Domain already in use"); + } + } + + if (removedIds.size > 0) { + await tx + .delete(servicePorts) + .where( + and( + eq(servicePorts.serviceId, serviceId), + inArray(servicePorts.id, [...removedIds]), + ), + ); + } + if (additions.length > 0) { + await tx.insert(servicePorts).values(additions); + } + }); + } catch (error) { + if ( + (error as { code?: string; constraint?: string }).code === "23505" && + (error as { constraint?: string }).constraint === + "service_ports_domain_unique" + ) { + throw new Error("Domain already in use"); } + throw error; } } @@ -1177,10 +1232,6 @@ export async function updateServiceConfig( if (!currentService) throw new Error("Service not found"); if (placement.mode === "automatic") { - if (currentService.serverlessEnabled) - throw new Error( - "Automatic placement is not supported for serverless services", - ); const volume = await tx .select({ id: serviceVolumes.id }) .from(serviceVolumes) diff --git a/web/components/service/details/networking-section.tsx b/web/components/service/details/networking-section.tsx index 7d202c87..69e72424 100644 --- a/web/components/service/details/networking-section.tsx +++ b/web/components/service/details/networking-section.tsx @@ -124,7 +124,7 @@ export const NetworkingSection = memo(function NetworkingSection({ port <= 65535; const canAdd = isValidPort && - !httpPorts.some((p) => p.port === port) && + !httpPorts.some((p) => p.port === port && !p.isPublic) && isValidDomain && !httpPorts.some((p) => p.domain === normalizedDomain) && !isSaving; diff --git a/web/components/service/details/replicas-section.tsx b/web/components/service/details/replicas-section.tsx index 280ddee3..ce54091d 100644 --- a/web/components/service/details/replicas-section.tsx +++ b/web/components/service/details/replicas-section.tsx @@ -53,7 +53,7 @@ export const ReplicasSection = memo(function ReplicasSection({ ); const [selectedServerId, setSelectedServerId] = useState(null); const [placementMode, setPlacementMode] = useState( - service.serverlessEnabled ? "manual" : service.placementMode, + service.placementMode, ); const [desiredReplicas, setDesiredReplicas] = useState(service.replicas); const [isEditing, setIsEditing] = useState(false); @@ -86,9 +86,7 @@ export const ReplicasSection = memo(function ReplicasSection({ } } setLocalReplicas(replicaMap); - setPlacementMode( - service.serverlessEnabled ? "manual" : service.placementMode, - ); + setPlacementMode(service.placementMode); setDesiredReplicas(service.replicas); } }, [ @@ -97,7 +95,6 @@ export const ReplicasSection = memo(function ReplicasSection({ service.stateful, service.lockedServerId, service.placementMode, - service.serverlessEnabled, service.replicas, isEditing, ]); @@ -205,7 +202,6 @@ export const ReplicasSection = memo(function ReplicasSection({ const handleModeChange = (mode: string) => { const nextMode = mode as PlacementMode; - if (nextMode === "automatic" && service.serverlessEnabled) return; if (nextMode === placementMode) return; setIsEditing(true); if (nextMode === "automatic") { @@ -384,11 +380,9 @@ export const ReplicasSection = memo(function ReplicasSection({
- {!service.serverlessEnabled && ( - - Automatic - - )} + + Automatic + Manual @@ -403,7 +397,8 @@ export const ReplicasSection = memo(function ReplicasSection({

The control plane distributes replicas evenly across healthy - nodes and moves them after failures. + {service.serverlessEnabled ? " proxy nodes" : " nodes"} and + moves them after failures.

replica.count > 0 && !replica.serverIsProxy, ); - const unavailableReason = - service.placementMode === "automatic" - ? "Switch to manual placement before enabling serverless" - : !hasPublicHttpEndpoint - ? "Add a public HTTP port with a domain to enable serverless" - : hasWorkerReplica - ? "Serverless services can only be deployed to proxy nodes" - : null; + const unavailableReason = !hasPublicHttpEndpoint + ? "Add a public HTTP port with a domain to enable serverless" + : hasWorkerReplica + ? "Serverless services can only be deployed to proxy nodes" + : null; const optionsDisabled = !!unavailableReason || isSaving; const parsed = useMemo( diff --git a/web/components/service/details/service-details-overview.tsx b/web/components/service/details/service-details-overview.tsx index bcfb4e5f..8d01f750 100644 --- a/web/components/service/details/service-details-overview.tsx +++ b/web/components/service/details/service-details-overview.tsx @@ -433,7 +433,7 @@ function ServiceConfigPanel({ service: Service; overview: OverviewData; }) { - const primaryEndpoint = getPrimaryEndpoint(overview.endpoints); + const displayedEndpoints = getDisplayedEndpoints(overview.endpoints); const statusClasses = STATUS_TONE_CLASSES[overview.status.tone]; const hasResourceLimits = service.resourceCpuLimit != null || service.resourceMemoryLimitMb != null; @@ -499,13 +499,20 @@ function ServiceConfigPanel({
- - + + {displayedEndpoints.map((endpoint, index) => ( + + {index > 0 ? ", " : null} + + + ))} {formatPortSummary(service.ports || [])} - {primaryEndpoint.kind !== "private" ? ( + {displayedEndpoints[0]?.kind !== "private" ? ( {`${service.hostname || service.name}.internal`} @@ -896,18 +903,28 @@ function formatInstanceSummary(overview: OverviewData): string { return `${overview.runningDeployments}/${configured} running`; } -function getPrimaryEndpoint(endpoints: EndpointItem[]): EndpointItem { - return ( - endpoints.find((endpoint) => endpoint.kind === "public") ?? - endpoints.find((endpoint) => endpoint.kind === "tcp") ?? +function getDisplayedEndpoints(endpoints: EndpointItem[]): EndpointItem[] { + const publicEndpoints = endpoints.filter( + (endpoint) => endpoint.kind === "public", + ); + if (publicEndpoints.length > 0) { + return publicEndpoints.sort((a, b) => a.label.localeCompare(b.label)); + } + + const tcpEndpoints = endpoints.filter((endpoint) => endpoint.kind === "tcp"); + if (tcpEndpoints.length > 0) { + return tcpEndpoints.sort((a, b) => a.label.localeCompare(b.label)); + } + + return [ endpoints[0] ?? { key: "none", kind: "private", typeLabel: "Private", label: "No endpoint", target: "Internal DNS", - } - ); + }, + ]; } function formatPortSummary(ports: Service["ports"]): string { diff --git a/web/db/schema.ts b/web/db/schema.ts index d91c35da..f7e4ab86 100644 --- a/web/db/schema.ts +++ b/web/db/schema.ts @@ -504,7 +504,7 @@ export const servicePorts = pgTable( .references(() => services.id, { onDelete: "cascade" }), port: integer("port").notNull(), isPublic: boolean("is_public").notNull().default(false), - domain: text("domain").unique(), + domain: text("domain").unique("service_ports_domain_unique"), protocol: text("protocol", { enum: ["http", "tcp", "udp"] }) .notNull() .default("http"), diff --git a/web/lib/agent/expected-state.ts b/web/lib/agent/expected-state.ts index db0ada54..fc05f4d5 100644 --- a/web/lib/agent/expected-state.ts +++ b/web/lib/agent/expected-state.ts @@ -17,9 +17,10 @@ import { } from "@/lib/deployment-status"; import { selectRoutingSyncRolloutIds } from "@/lib/routing-sync"; import { parseServiceRevisionSpec } from "@/lib/service-revision-changes"; -import type { - ServiceRevisionSecret, - ServiceRevisionSpec, +import { + getPublishedContainerPorts, + type ServiceRevisionSecret, + type ServiceRevisionSpec, } from "@/lib/service-revision-spec"; import { getWireGuardPeers } from "@/lib/wireguard"; @@ -269,7 +270,9 @@ async function buildExpectedContainers( ), ); - const serviceIds = [...new Set(serverDeployments.map((dep) => dep.serviceId))]; + const serviceIds = [ + ...new Set(serverDeployments.map((dep) => dep.serviceId)), + ]; const revisionIds = [ ...new Set(serverDeployments.map((dep) => dep.serviceRevisionId)), ]; @@ -361,10 +364,12 @@ export function buildExpectedContainersFromRows({ containerPort: port.containerPort, hostPort: port.hostPort, })); - const expectedContainerPorts = specification.ports - .map((port) => port.containerPort) - .sort((a, b) => a - b); - const allocatedContainerPorts = ports.map((port) => port.containerPort); + const expectedContainerPorts = getPublishedContainerPorts( + specification.ports, + ); + const allocatedContainerPorts = [ + ...new Set(ports.map((port) => port.containerPort)), + ]; if ( JSON.stringify(expectedContainerPorts) !== JSON.stringify(allocatedContainerPorts) @@ -855,6 +860,8 @@ function compareServicePorts(a: RouteServicePort, b: RouteServicePort) { return ( a.serviceId.localeCompare(b.serviceId) || a.protocol.localeCompare(b.protocol) || - a.port - b.port + a.port - b.port || + (a.domain ?? "").localeCompare(b.domain ?? "") || + a.id.localeCompare(b.id) ); } diff --git a/web/lib/inngest/functions/rollout-helpers.ts b/web/lib/inngest/functions/rollout-helpers.ts index 0b7e153a..79d30979 100644 --- a/web/lib/inngest/functions/rollout-helpers.ts +++ b/web/lib/inngest/functions/rollout-helpers.ts @@ -9,7 +9,10 @@ import { services, } from "@/db/schema"; import { getCertificate, issueCertificate } from "@/lib/acme-manager"; -import type { ServiceRevisionSpec } from "@/lib/service-revision-spec"; +import { + getPublishedContainerPorts, + type ServiceRevisionSpec, +} from "@/lib/service-revision-spec"; import { assignContainerIp } from "@/lib/wireguard"; import { enqueueWork } from "@/lib/work-queue"; @@ -201,6 +204,7 @@ export async function resolveRevisionPlacements( export async function validateServers( placements: Placement[], + requireProxy = false, ): Promise< Map< string, @@ -231,6 +235,9 @@ export async function validateServers( if (!server.wireguardIp) { throw new Error(`Server ${server.name} has no WireGuard IP`); } + if (requireProxy && !server.isProxy) { + throw new Error(`Server ${server.name} is not a proxy node`); + } } } @@ -381,6 +388,9 @@ export async function createDeploymentRecords( } const deploymentIds = existingDeployments.map((deployment) => deployment.id); + const publishedContainerPorts = getPublishedContainerPorts( + specification.ports, + ); for (const placement of placements) { if (placement.replicas <= 0) continue; @@ -396,7 +406,7 @@ export async function createDeploymentRecords( const deploymentId = randomUUID(); const hostPorts = await allocateHostPorts( server.id, - specification.ports.length, + publishedContainerPorts.length, ); const ipAddress = await assignContainerIp(server.id); @@ -425,12 +435,12 @@ export async function createDeploymentRecords( rolloutId, }); - if (specification.ports.length > 0) { + if (publishedContainerPorts.length > 0) { await tx.insert(deploymentPorts).values( - specification.ports.map((port, index) => ({ + publishedContainerPorts.map((containerPort, index) => ({ id: randomUUID(), deploymentId, - containerPort: port.containerPort, + containerPort, hostPort: hostPorts[index], })), ); diff --git a/web/lib/inngest/functions/rollout-workflow.ts b/web/lib/inngest/functions/rollout-workflow.ts index bc835531..7316bed3 100644 --- a/web/lib/inngest/functions/rollout-workflow.ts +++ b/web/lib/inngest/functions/rollout-workflow.ts @@ -314,7 +314,10 @@ export const rolloutWorkflow = inngest.createFunction( const serverValidation = await step.run("validate-servers", async () => { try { - const serverMap = await validateServers(placements); + const serverMap = await validateServers( + placements, + specification.serverless.enabled, + ); const ids = [...serverMap.keys()]; await ingestRolloutLog( rolloutId, @@ -422,7 +425,10 @@ export const rolloutWorkflow = inngest.createFunction( .set({ currentStage: "deploying" }) .where(eq(rollouts.id, rolloutId)); - const serverMap = await validateServers(placements); + const serverMap = await validateServers( + placements, + specification.serverless.enabled, + ); const result = await createDeploymentRecords(rolloutId, serviceId, { revisionId: revision.id, diff --git a/web/lib/port-allocation.ts b/web/lib/port-allocation.ts index 19535930..f54eb773 100644 --- a/web/lib/port-allocation.ts +++ b/web/lib/port-allocation.ts @@ -1,17 +1,25 @@ -import { db } from "@/db"; -import { servicePorts } from "@/db/schema"; import { asc, eq } from "drizzle-orm"; +import type { db } from "@/db"; +import { servicePorts } from "@/db/schema"; const TCP_PORT_START = 10000; const TCP_PORT_END = 10999; const UDP_PORT_START = 11000; const UDP_PORT_END = 11999; -export async function allocatePort(protocol: "tcp" | "udp"): Promise { +type PortAllocationTransaction = Parameters< + Parameters[0] +>[0]; + +export async function allocatePort( + tx: PortAllocationTransaction, + protocol: "tcp" | "udp", + reserved: ReadonlySet = new Set(), +): Promise { const portStart = protocol === "tcp" ? TCP_PORT_START : UDP_PORT_START; const portEnd = protocol === "tcp" ? TCP_PORT_END : UDP_PORT_END; - const usedPorts = await db + const usedPorts = await tx .select({ port: servicePorts.externalPort }) .from(servicePorts) .where(eq(servicePorts.protocol, protocol)) @@ -22,7 +30,7 @@ export async function allocatePort(protocol: "tcp" | "udp"): Promise { ); for (let port = portStart; port <= portEnd; port++) { - if (!usedSet.has(port)) { + if (!usedSet.has(port) && !reserved.has(port)) { return port; } } diff --git a/web/lib/public-api.ts b/web/lib/public-api.ts index 5f80af4f..0d722a85 100644 --- a/web/lib/public-api.ts +++ b/web/lib/public-api.ts @@ -18,6 +18,7 @@ import { validateDockerImageInternal } from "@/lib/docker-image"; import { getServiceTotalReplicas } from "@/lib/service-config"; import { parseServiceRevisionSpec } from "@/lib/service-revision-changes"; import { + findServicePortValidationIssue, getDefaultServiceHostname, getServiceRevisionTotalReplicas, } from "@/lib/service-revision-spec"; @@ -689,10 +690,10 @@ export async function patchConfiguration( const source = resolvePersistedSourceFromRows(persisted, repo); if ( input.placement?.mode === "automatic" && - (persisted.stateful || persisted.serverlessEnabled || volumes.length > 0) + (persisted.stateful || volumes.length > 0) ) { domainError( - "Automatic placement is not supported for stateful, serverless, or volume-backed services", + "Automatic placement is not supported for stateful or volume-backed services", "AUTOMATIC_PLACEMENT_UNSUPPORTED", 400, ); @@ -795,18 +796,20 @@ export async function patchConfiguration( 400, ); } - if ( - new Set(input.ports.map((port) => port.containerPort)).size !== - input.ports.length - ) { - domainError("Port numbers must be unique", "DUPLICATE_PORT", 400); + const portIssue = findServicePortValidationIssue( + input.ports.map((port) => ({ + containerPort: port.containerPort, + isPublic: port.public, + domain: port.domain ?? null, + protocol: "http" as const, + })), + ); + if (portIssue) { + domainError(portIssue.message, portIssue.code, 400); } const domains = input.ports.flatMap((port) => port.public && port.domain ? [port.domain] : [], ); - if (new Set(domains).size !== domains.length) { - domainError("Port domains must be unique", "DUPLICATE_DOMAIN", 400); - } for (const domain of domains) { const duplicate = await tx .select({ id: servicePorts.id }) @@ -965,28 +968,50 @@ export async function patchConfiguration( if (input.ports) { const currentPorts = ports .map((port) => [port.port, port.isPublic, port.domain] as const) - .toSorted((a, b) => a[0] - b[0]); + .toSorted( + (a, b) => + a[0] - b[0] || + Number(a[1]) - Number(b[1]) || + (a[2] ?? "").localeCompare(b[2] ?? ""), + ); const desiredPorts = input.ports .map( (port) => [port.containerPort, port.public, port.domain ?? null] as const, ) - .toSorted((a, b) => a[0] - b[0]); + .toSorted( + (a, b) => + a[0] - b[0] || + Number(a[1]) - Number(b[1]) || + (a[2] ?? "").localeCompare(b[2] ?? ""), + ); if (changed("ports", currentPorts, desiredPorts)) { await tx .delete(servicePorts) .where(eq(servicePorts.serviceId, service.id)); if (input.ports.length > 0) { - await tx.insert(servicePorts).values( - input.ports.map((port) => ({ - id: randomUUID(), - serviceId: service.id, - port: port.containerPort, - isPublic: port.public, - domain: port.public ? (port.domain ?? null) : null, - protocol: "http" as const, - })), - ); + try { + await tx.insert(servicePorts).values( + input.ports.map((port) => ({ + id: randomUUID(), + serviceId: service.id, + port: port.containerPort, + isPublic: port.public, + domain: port.public ? (port.domain ?? null) : null, + protocol: "http" as const, + })), + ); + } catch (error) { + if ( + (error as { code?: string; constraint?: string }).code === + "23505" && + (error as { constraint?: string }).constraint === + "service_ports_domain_unique" + ) { + domainError("Port domain is already in use", "DOMAIN_CONFLICT"); + } + throw error; + } } } } diff --git a/web/lib/scheduler.ts b/web/lib/scheduler.ts index b66b76fb..e1cc1f2d 100644 --- a/web/lib/scheduler.ts +++ b/web/lib/scheduler.ts @@ -199,6 +199,7 @@ export async function recoverInvalidAutomaticPlacements( specification: serviceRevisions.specification, serverStatus: servers.status, serverWireguardIp: servers.wireguardIp, + serverIsProxy: servers.isProxy, lastRecoveryAttemptAt: services.lastAutomaticRecoveryAttemptAt, }) .from(deployments) @@ -262,7 +263,9 @@ export async function recoverInvalidAutomaticPlacements( continue; const hasInvalidPlacement = serviceDeployments.some( (deployment) => - deployment.serverStatus !== "online" || !deployment.serverWireguardIp, + deployment.serverStatus !== "online" || + !deployment.serverWireguardIp || + (specification.serverless.enabled && !deployment.serverIsProxy), ); if (!hasInvalidPlacement) continue; diff --git a/web/lib/service-config.ts b/web/lib/service-config.ts index 51394867..f7f30dd8 100644 --- a/web/lib/service-config.ts +++ b/web/lib/service-config.ts @@ -538,49 +538,20 @@ export function diffConfigs( }); } - const deployedPortsMap = new Map( - (deployed.ports || []).map((p) => [p.port, p]), - ); - const currentPortsMap = new Map( - (current.ports || []).map((p) => [p.port, p]), - ); - - for (const [port, currentPort] of currentPortsMap) { - const deployedPort = deployedPortsMap.get(port); - const portType = currentPort.isPublic ? "public" : "internal"; - const portDesc = currentPort.domain - ? `${portType}, ${currentPort.domain}` - : portType; - - if (!deployedPort) { + const deployedPortsByNumber = groupPortsByNumber(deployed.ports || []); + const currentPortsByNumber = groupPortsByNumber(current.ports || []); + const portNumbers = Array.from( + new Set([...deployedPortsByNumber.keys(), ...currentPortsByNumber.keys()]), + ).sort((a, b) => a - b); + + for (const port of portNumbers) { + const deployedDesc = describePorts(deployedPortsByNumber.get(port) || []); + const currentDesc = describePorts(currentPortsByNumber.get(port) || []); + if (deployedDesc !== currentDesc) { changes.push({ field: `Port ${port}`, - from: "(none)", - to: portDesc, - }); - } else { - const deployedType = deployedPort.isPublic ? "public" : "internal"; - const deployedDesc = deployedPort.domain - ? `${deployedType}, ${deployedPort.domain}` - : deployedType; - - if (deployedDesc !== portDesc) { - changes.push({ - field: `Port ${port}`, - from: deployedDesc, - to: portDesc, - }); - } - } - } - - for (const [port, deployedPort] of deployedPortsMap) { - if (!currentPortsMap.has(port)) { - const deployedType = deployedPort.isPublic ? "public" : "internal"; - changes.push({ - field: `Port ${port}`, - from: deployedType, - to: "(removed)", + from: deployedDesc || "(none)", + to: currentDesc || "(removed)", }); } } @@ -659,6 +630,30 @@ export function diffConfigs( return changes; } +function groupPortsByNumber(ports: PortConfig[]): Map { + const grouped = new Map(); + for (const port of ports) { + const entries = grouped.get(port.port) || []; + entries.push(port); + grouped.set(port.port, entries); + } + return grouped; +} + +function describePorts(ports: PortConfig[]): string { + return ports + .map((port) => { + const details = [port.isPublic ? "public" : "internal"]; + const protocol = port.protocol || "http"; + if (protocol !== "http") details.push(protocol.toUpperCase()); + if (port.domain) details.push(port.domain); + if (port.tlsPassthrough) details.push("TLS passthrough"); + return details.join(", "); + }) + .sort((a, b) => a.localeCompare(b)) + .join("; "); +} + export function normalizeServerlessConfig( config: ServerlessConfig | undefined, ): ServerlessConfig { diff --git a/web/lib/service-revision-changes.ts b/web/lib/service-revision-changes.ts index 21cd9000..cdb0448c 100644 --- a/web/lib/service-revision-changes.ts +++ b/web/lib/service-revision-changes.ts @@ -6,6 +6,7 @@ import type { ServiceRevisionPort, ServiceRevisionSpec, } from "@/lib/service-revision-spec"; +import { validateServiceRevisionPorts } from "@/lib/service-revision-spec"; const serviceRevisionSpecFields = { image: z.string(), @@ -105,11 +106,6 @@ const serviceRevisionSpecSchema = z code: "custom", message: "Stateful services cannot use automatic placement", }); - if (spec.serverless.enabled && spec.placement.mode === "automatic") - context.addIssue({ - code: "custom", - message: "Serverless services cannot use automatic placement", - }); }); export type ServiceRevisionChange = { @@ -143,9 +139,19 @@ export function parseServiceRevisionSpec(value: unknown): ServiceRevisionSpec { const version = (value as { schemaVersion?: unknown } | null)?.schemaVersion; if (version === 2) { const legacy = serviceRevisionSpecV2Schema.parse(value); - return { ...legacy, schemaVersion: 3, placement: { mode: "manual" } }; + const specification = { + ...legacy, + schemaVersion: 3 as const, + placement: { mode: "manual" as const }, + }; + validateServiceRevisionPorts(specification.ports); + return specification; } - return serviceRevisionSpecSchema.parse(value) as ServiceRevisionSpec; + const specification = serviceRevisionSpecSchema.parse( + value, + ) as ServiceRevisionSpec; + validateServiceRevisionPorts(specification.ports); + return specification; } function compareStrings(a: string, b: string): number { diff --git a/web/lib/service-revision-spec.ts b/web/lib/service-revision-spec.ts index 5b289cb4..c14f461e 100644 --- a/web/lib/service-revision-spec.ts +++ b/web/lib/service-revision-spec.ts @@ -33,6 +33,83 @@ export type ServiceRevisionPort = { tlsPassthrough: boolean; }; +export type ServicePortValidationIssue = { + code: "DUPLICATE_DOMAIN" | "DUPLICATE_PORT" | "INVALID_DOMAIN"; + message: string; +}; + +export function findServicePortValidationIssue( + ports: Pick< + ServiceRevisionPort, + "containerPort" | "isPublic" | "domain" | "protocol" + >[], +): ServicePortValidationIssue | null { + const domains = new Set(); + const portsByNumberAndProtocol = Map.groupBy( + ports, + (port) => `${port.containerPort}/${port.protocol}`, + ); + + for (const port of ports) { + if (port.protocol === "http" && port.isPublic && !port.domain) { + return { + code: "INVALID_DOMAIN", + message: "Public HTTP ports require a domain", + }; + } + if ((!port.isPublic || port.protocol !== "http") && port.domain) { + return { + code: "INVALID_DOMAIN", + message: "Only public HTTP ports can define a domain", + }; + } + if (port.domain) { + const domain = port.domain.toLowerCase(); + if (domains.has(domain)) { + return { + code: "DUPLICATE_DOMAIN", + message: "Port domains must be unique", + }; + } + domains.add(domain); + } + } + + for (const group of portsByNumberAndProtocol.values()) { + if ( + group.length > 1 && + !group.every( + (port) => port.protocol === "http" && port.isPublic && port.domain, + ) + ) { + return { + code: "DUPLICATE_PORT", + message: `Port ${group[0].containerPort} (${group[0].protocol}) can only be repeated for public HTTP domains`, + }; + } + } + + return null; +} + +export function validateServiceRevisionPorts( + ports: Pick< + ServiceRevisionPort, + "containerPort" | "isPublic" | "domain" | "protocol" + >[], +) { + const issue = findServicePortValidationIssue(ports); + if (issue) throw new Error(issue.message); +} + +export function getPublishedContainerPorts( + ports: Pick[], +): number[] { + return [...new Set(ports.map((port) => port.containerPort))].sort( + (a, b) => a - b, + ); +} + export type ServiceRevisionSecret = { key: string; encryptedValue: string; @@ -143,6 +220,7 @@ function validateServiceRevisionSpec( specification: ServiceRevisionSpec, allowNoPlacements: boolean, ) { + validateServiceRevisionPorts(specification.ports); const totalReplicas = getServiceRevisionTotalReplicas(specification); if (totalReplicas < 1 && !allowNoPlacements) { @@ -166,12 +244,6 @@ function validateServiceRevisionSpec( if (specification.stateful && specification.placement.mode === "automatic") { throw new Error("Stateful services cannot use automatic placement"); } - if ( - specification.serverless.enabled && - specification.placement.mode === "automatic" - ) { - throw new Error("Serverless services cannot use automatic placement"); - } if (specification.stateful && totalReplicas !== 1) { throw new Error("Stateful services can only have exactly 1 replica"); } diff --git a/web/lib/victoria-metrics.ts b/web/lib/victoria-metrics.ts index 301b1773..1b25242b 100644 --- a/web/lib/victoria-metrics.ts +++ b/web/lib/victoria-metrics.ts @@ -246,11 +246,10 @@ export async function queryServiceMetrics(options: { const window = getMetricWindow(options.range, now); if (!endpoint) return createEmptyServiceMetrics(options.range, now); - const serviceMatcher = buildTraefikServiceMatcher(options.serviceId); const serviceId = escapePromQL(options.serviceId); const rangeWindow = formatPromDuration(window.stepSeconds); const totalWindow = formatPromDuration(window.durationMs / 1000); - const traefikFilter = `service=~"${serviceMatcher}"`; + const traefikFilter = `service_id="${serviceId}"`; const queryStart = addMilliseconds( window.start, window.stepSeconds * SECOND_IN_MILLISECONDS, @@ -642,10 +641,6 @@ export function getMetricWindow( }; } -export function buildTraefikServiceMatcher(serviceId: string): string { - return `^${escapePromRegex(serviceId)}(@file)?$`; -} - export function formatPromDuration(seconds: number): string { if (seconds % 86400 === 0) return `${seconds / 86400}d`; if (seconds % 3600 === 0) return `${seconds / 3600}h`; @@ -755,7 +750,3 @@ function escapePromQL(value: string) { .replace(/"/g, '\\"') .replace(/\n/g, "\\n"); } - -function escapePromRegex(value: string) { - return value.replace(/[\\^$.*+?()[\]{}|]/g, "\\$&"); -} diff --git a/web/tests/expected-state.test.ts b/web/tests/expected-state.test.ts index 138355a0..86cff6ef 100644 --- a/web/tests/expected-state.test.ts +++ b/web/tests/expected-state.test.ts @@ -259,6 +259,108 @@ describe("expected-state pure builders", () => { ).toThrow("Deployment dep_incomplete_ports has incomplete port allocation"); }); + it("publishes one physical port for multiple HTTP domain aliases", () => { + const ports = ["app.example.com", "www.example.com"].map((domain) => ({ + containerPort: 3000, + isPublic: true, + domain, + protocol: "http" as const, + externalPort: null, + tlsPassthrough: false, + })); + const containers = buildExpectedContainersFromRows({ + deployments: [ + { + id: "dep_aliases", + serviceId: "svc_aliases", + serviceRevisionId: "rev_svc_aliases", + runtimeDesiredState: "running", + }, + ] as any, + services: [{ id: "svc_aliases", name: "aliases" }] as any, + revisions: [revision("svc_aliases", { ports })], + deploymentPorts: [ + { deploymentId: "dep_aliases", containerPort: 3000, hostPort: 31000 }, + ] as any, + }); + + expect(containers[0]?.ports).toEqual([ + { containerPort: 3000, hostPort: 31000 }, + ]); + }); + + it("preserves historical physical mappings for cross-protocol ports", () => { + const containers = buildExpectedContainersFromRows({ + deployments: [ + { + id: "dep_protocols", + serviceId: "svc_protocols", + serviceRevisionId: "rev_svc_protocols", + runtimeDesiredState: "running", + }, + ] as any, + services: [{ id: "svc_protocols", name: "protocols" }] as any, + revisions: [ + revision("svc_protocols", { + ports: [ + { + containerPort: 3000, + isPublic: true, + domain: "app.example.com", + protocol: "http", + externalPort: null, + tlsPassthrough: false, + }, + { + containerPort: 3000, + isPublic: true, + domain: null, + protocol: "tcp", + externalPort: 10000, + tlsPassthrough: false, + }, + ], + }), + ], + deploymentPorts: [ + { deploymentId: "dep_protocols", containerPort: 3000, hostPort: 31000 }, + { deploymentId: "dep_protocols", containerPort: 3000, hostPort: 31001 }, + ] as any, + }); + + expect(containers[0]?.ports).toEqual([ + { containerPort: 3000, hostPort: 31000 }, + { containerPort: 3000, hostPort: 31001 }, + ]); + }); + + it("builds one HTTP route per domain alias", () => { + const routes = buildTraefikRoutes({ + serverId: "server_1", + ports: ["app.example.com", "www.example.com"].map((domain, index) => ({ + id: `port_${index}`, + serviceId: "svc_1", + port: 3000, + isPublic: true, + protocol: "http", + domain, + })) as any, + routableDeployments: [ + { serviceId: "svc_1", serverId: "server_1", ipAddress: "10.0.0.1" }, + ] as any, + }); + + expect(routes.httpRoutes.map((route) => route.domain)).toEqual([ + "app.example.com", + "www.example.com", + ]); + expect( + routes.httpRoutes.every( + (route) => route.upstreams[0]?.url === "10.0.0.1:3000", + ), + ).toBe(true); + }); + it("keeps HTTP local upstreams before remote upstreams", () => { const routes = buildTraefikRoutes({ serverId: "server_local", @@ -870,6 +972,14 @@ describe("expected-state pure builders", () => { externalPort: null, tlsPassthrough: false, }, + { + containerPort: 3000, + isPublic: true, + domain: "www.sleepy.example.com", + protocol: "http", + externalPort: null, + tlsPassthrough: false, + }, ], }); const routes = buildServerlessRoutesFromRows({ @@ -903,6 +1013,15 @@ describe("expected-state pure builders", () => { localDeploymentIds: ["dep_sleeping"], upstreams: [], }, + { + serviceId: "svc_1", + domain: "www.sleepy.example.com", + port: 3000, + sleepAfterSeconds: 300, + wakeTimeoutSeconds: 120, + localDeploymentIds: ["dep_sleeping"], + upstreams: [], + }, ]); }); diff --git a/web/tests/service-config.test.ts b/web/tests/service-config.test.ts index 95933302..db76a47b 100644 --- a/web/tests/service-config.test.ts +++ b/web/tests/service-config.test.ts @@ -234,6 +234,52 @@ describe("service config", () => { expect(diffConfigs(deployed, current)).toEqual([]); }); + it("compares multiple domains on one port without depending on order", () => { + const first = { + port: 8080, + isPublic: true, + domain: "app.example.com", + protocol: "http" as const, + }; + const second = { + port: 8080, + isPublic: true, + domain: "www.example.com", + protocol: "http" as const, + }; + + expect( + diffConfigs( + deployedConfig({ ports: [first, second] }), + deployedConfig({ ports: [second, first] }), + ), + ).toEqual([]); + }); + + it("reports changes to one of multiple domains on a shared port", () => { + const unchanged = { + port: 8080, + isPublic: true, + domain: "app.example.com", + protocol: "http" as const, + }; + + expect( + diffConfigs( + deployedConfig({ + ports: [unchanged, { ...unchanged, domain: "old.example.com" }], + }), + deployedConfig({ + ports: [{ ...unchanged, domain: "new.example.com" }, unchanged], + }), + ), + ).toContainEqual({ + field: "Port 8080", + from: "public, app.example.com; public, old.example.com", + to: "public, app.example.com; public, new.example.com", + }); + }); + it("reports serverless changes as pending config", () => { const changes = diffConfigs(deployedConfig(), { source: { type: "image", image: "nginx" }, diff --git a/web/tests/service-revision-changes.test.ts b/web/tests/service-revision-changes.test.ts index 6304b160..4839a784 100644 --- a/web/tests/service-revision-changes.test.ts +++ b/web/tests/service-revision-changes.test.ts @@ -147,15 +147,46 @@ describe("diffServiceRevisionSpecs", () => { ]); }); - it("rejects persisted automatic serverless revisions", () => { + it("accepts persisted automatic serverless revisions", () => { const automatic = spec(); automatic.placement = { mode: "automatic", replicas: 1 }; automatic.placements = []; automatic.serverless.enabled = true; + automatic.volumes = []; + automatic.ports[0] = { + containerPort: 3000, + isPublic: true, + domain: "app.example.com", + protocol: "http", + externalPort: null, + tlsPassthrough: false, + }; - expect(() => parseServiceRevisionSpec(automatic)).toThrow( - "Serverless services cannot use automatic placement", - ); + expect(parseServiceRevisionSpec(automatic)).toMatchObject({ + serverless: { enabled: true }, + placement: { mode: "automatic", replicas: 1 }, + volumes: [], + }); + }); + + it("accepts persisted stateful and volume-backed serverless revisions", () => { + const stateful = spec(); + stateful.serverless.enabled = true; + stateful.stateful = true; + + expect(parseServiceRevisionSpec(stateful)).toMatchObject({ + stateful: true, + serverless: { enabled: true }, + volumes: [{ name: "data", containerPath: "/data" }], + }); + + const volumeBacked = spec(); + volumeBacked.serverless.enabled = true; + + expect(parseServiceRevisionSpec(volumeBacked)).toMatchObject({ + serverless: { enabled: true }, + volumes: [{ name: "data", containerPath: "/data" }], + }); }); it("never exposes secret ciphertext while detecting additions, updates, and removals", () => { diff --git a/web/tests/service-revision-spec.test.ts b/web/tests/service-revision-spec.test.ts index 1bf4831b..a64d9a63 100644 --- a/web/tests/service-revision-spec.test.ts +++ b/web/tests/service-revision-spec.test.ts @@ -33,7 +33,7 @@ function draft( { port: 443, isPublic: true, - domain: "api.example.com", + domain: null, protocol: "tcp", externalPort: 443, tlsPassthrough: true, @@ -85,6 +85,7 @@ describe("service revision specification", () => { input.service.serverlessSleepAfterSeconds = 30; input.service.healthCheckInterval = null; input.ports[0].protocol = null; + input.ports[0].domain = "api.example.com"; input.ports[0].tlsPassthrough = null; const spec = buildServiceRevisionSpec(input); @@ -105,6 +106,83 @@ describe("service revision specification", () => { ); }); + it("allows distinct public HTTP domains to share a container port", () => { + const input = draft({ + ports: [ + { + port: 3000, + isPublic: true, + domain: "app.example.com", + protocol: "http", + externalPort: null, + tlsPassthrough: false, + }, + { + port: 3000, + isPublic: true, + domain: "www.example.com", + protocol: "http", + externalPort: null, + tlsPassthrough: false, + }, + ], + }); + + expect(buildServiceRevisionSpec(input).ports).toHaveLength(2); + }); + + it("allows different protocols to share a numeric container port", () => { + const input = draft({ + ports: [ + { + port: 3000, + isPublic: true, + domain: "app.example.com", + protocol: "http", + externalPort: null, + tlsPassthrough: false, + }, + { + port: 3000, + isPublic: true, + domain: null, + protocol: "tcp", + externalPort: 10000, + tlsPassthrough: false, + }, + ], + }); + + expect(buildServiceRevisionSpec(input).ports).toHaveLength(2); + }); + + it("rejects repeated container ports that are not HTTP aliases", () => { + const input = draft({ + ports: [ + { + port: 3000, + isPublic: true, + domain: "app.example.com", + protocol: "http", + externalPort: null, + tlsPassthrough: false, + }, + { + port: 3000, + isPublic: false, + domain: null, + protocol: "http", + externalPort: null, + tlsPassthrough: false, + }, + ], + }); + + expect(() => buildServiceRevisionSpec(input)).toThrow( + "can only be repeated for public HTTP domains", + ); + }); + it("snapshots GitHub source provenance and the reserved runtime image", () => { const spec = buildServiceRevisionSpec(draft(), { image: "registry.test/project/service:revision-1", @@ -152,7 +230,7 @@ describe("service revision specification", () => { }); }); - it("rejects automatic placement for stateful, volume-backed, and serverless services", () => { + it("rejects automatic placement for stateful and volume-backed services", () => { const stateful = draft({ volumes: [] }); stateful.service.stateful = true; stateful.service.placementMode = "automatic"; @@ -169,19 +247,38 @@ describe("service revision specification", () => { expect(() => buildServiceRevisionSpec(volumeBacked)).toThrow( "Services with volumes cannot use automatic placement", ); + }); - const serverless = draft({ volumes: [] }); + it("allows automatic placement for stateless serverless services", () => { + const serverless = draft({ + volumes: [], + ports: [ + { + port: 3000, + isPublic: true, + domain: "api.example.com", + protocol: "http", + externalPort: null, + tlsPassthrough: false, + }, + ], + }); serverless.service.serverlessEnabled = true; serverless.service.placementMode = "automatic"; - serverless.service.replicas = 1; + serverless.service.replicas = 3; - expect(() => buildServiceRevisionSpec(serverless)).toThrow( - "Serverless services cannot use automatic placement", - ); + expect(buildServiceRevisionSpec(serverless)).toMatchObject({ + stateful: false, + serverless: { enabled: true }, + placement: { mode: "automatic", replicas: 3 }, + placements: [], + volumes: [], + }); }); it("rejects serverless revisions without a public HTTP port and domain", () => { const input = draft({ + volumes: [], ports: [ { port: 3000, @@ -214,4 +311,29 @@ describe("service revision specification", () => { expect(() => buildServiceRevisionSpec(input)).not.toThrow(); }); + + it("accepts manually placed stateful serverless revisions", () => { + const input = draft({ + placements: [{ serverId: "proxy-server", count: 1 }], + volumes: [{ name: "data", containerPath: "/data" }], + }); + input.service.stateful = true; + input.service.serverlessEnabled = true; + input.ports[0] = { + port: 443, + isPublic: true, + domain: "api.example.com", + protocol: "http", + externalPort: null, + tlsPassthrough: false, + }; + + expect(buildServiceRevisionSpec(input)).toMatchObject({ + stateful: true, + serverless: { enabled: true }, + placement: { mode: "manual" }, + placements: [{ serverId: "proxy-server", count: 1 }], + volumes: [{ name: "data", containerPath: "/data" }], + }); + }); }); diff --git a/web/tests/victoria-metrics-service-metrics.test.ts b/web/tests/victoria-metrics-service-metrics.test.ts index cf44d606..08201ca7 100644 --- a/web/tests/victoria-metrics-service-metrics.test.ts +++ b/web/tests/victoria-metrics-service-metrics.test.ts @@ -1,6 +1,5 @@ import { afterEach, describe, expect, it, vi } from "vitest"; import { - buildTraefikServiceMatcher, createEmptyServiceMetrics, formatPromDuration, queryServiceMetrics, @@ -23,13 +22,6 @@ describe("VictoriaMetrics service metrics", () => { expect(formatPromDuration(45)).toBe("45s"); }); - it("matches Traefik service labels with optional provider suffix", () => { - expect(buildTraefikServiceMatcher(SERVICE_ID)).toBe( - `^${SERVICE_ID}(@file)?$`, - ); - expect(buildTraefikServiceMatcher("svc.1")).toBe("^svc\\.1(@file)?$"); - }); - it("creates an empty metrics payload", () => { expect( createEmptyServiceMetrics("1h", new Date("2026-07-02T01:00:00Z")), @@ -129,7 +121,7 @@ describe("VictoriaMetrics service metrics", () => { expect(instantTimes).toEqual([String(END_TS), String(END_TS)]); expect(queries.some((query) => query.includes("LogSQL"))).toBe(false); expect(queries).toContain( - `sum by (code) (increase(traefik_service_requests_total{service=~"^${SERVICE_ID}(@file)?$"}[5m]))`, + `sum by (code) (increase(traefik_service_requests_total{service_id="${SERVICE_ID}"}[5m]))`, ); expect(queries).toContain( `sum(avg_over_time(techulus_service_cpu_usage_percent{service_id="${SERVICE_ID}"}[5m]))`, @@ -138,10 +130,10 @@ describe("VictoriaMetrics service metrics", () => { `sum(avg_over_time(techulus_service_memory_usage_percent{service_id="${SERVICE_ID}"}[5m]))`, ); expect(queries).toContain( - `sum(increase(traefik_service_requests_bytes_total{service=~"^${SERVICE_ID}(@file)?$"}[1d]))`, + `sum(increase(traefik_service_requests_bytes_total{service_id="${SERVICE_ID}"}[1d]))`, ); expect(queries).toContain( - `sum(increase(traefik_service_responses_bytes_total{service=~"^${SERVICE_ID}(@file)?$"}[1d]))`, + `sum(increase(traefik_service_responses_bytes_total{service_id="${SERVICE_ID}"}[1d]))`, ); expect( starts.every((start) => start === String(END_TS - 24 * 60 * 60 + 5 * 60)),