Add Traefik Hub Integration (Experimental Feature)
This commit is contained in:
Jean-Baptiste Doumenjou
GitHub
parent
8d58f33a28
commit
fbdb6e6e78
20 changed files with 992 additions and 11 deletions
146
pkg/provider/hub/handler.go
Normal file
146
pkg/provider/hub/handler.go
Normal file
|
|
@ -0,0 +1,146 @@
|
|||
package hub
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"fmt"
|
||||
"net/http"
|
||||
"net/url"
|
||||
"sync/atomic"
|
||||
|
||||
"github.com/traefik/traefik/v2/pkg/config/dynamic"
|
||||
"github.com/traefik/traefik/v2/pkg/log"
|
||||
)
|
||||
|
||||
type handler struct {
|
||||
mux *http.ServeMux
|
||||
|
||||
client http.Client
|
||||
|
||||
entryPoint string
|
||||
port int
|
||||
tlsCfg *TLS
|
||||
|
||||
// Accessed atomically.
|
||||
lastCfgUnixNano int64
|
||||
|
||||
cfgChan chan<- dynamic.Message
|
||||
}
|
||||
|
||||
func newHandler(entryPoint string, port int, cfgChan chan<- dynamic.Message, tlsCfg *TLS, client http.Client) http.Handler {
|
||||
h := &handler{
|
||||
mux: http.NewServeMux(),
|
||||
entryPoint: entryPoint,
|
||||
port: port,
|
||||
cfgChan: cfgChan,
|
||||
tlsCfg: tlsCfg,
|
||||
client: client,
|
||||
}
|
||||
|
||||
h.mux.HandleFunc("/config", h.handleConfig)
|
||||
h.mux.HandleFunc("/discover-ip", h.handleDiscoverIP)
|
||||
h.mux.HandleFunc("/state", h.handleState)
|
||||
|
||||
return h
|
||||
}
|
||||
|
||||
type configRequest struct {
|
||||
UnixNano int64 `json:"unixNano"`
|
||||
Configuration *dynamic.Configuration `json:"configuration"`
|
||||
}
|
||||
|
||||
func (h *handler) handleConfig(rw http.ResponseWriter, req *http.Request) {
|
||||
if req.Method != http.MethodPost {
|
||||
http.Error(rw, http.StatusText(http.StatusMethodNotAllowed), http.StatusMethodNotAllowed)
|
||||
return
|
||||
}
|
||||
|
||||
payload := &configRequest{Configuration: emptyDynamicConfiguration()}
|
||||
if err := json.NewDecoder(req.Body).Decode(payload); err != nil {
|
||||
err = fmt.Errorf("decoding config request: %w", err)
|
||||
log.WithoutContext().Errorf("Handling config: %v", err)
|
||||
http.Error(rw, http.StatusText(http.StatusBadRequest), http.StatusBadRequest)
|
||||
return
|
||||
}
|
||||
|
||||
cfg := payload.Configuration
|
||||
patchDynamicConfiguration(cfg, h.entryPoint, h.port, h.tlsCfg)
|
||||
|
||||
// We can safely drop messages here if the other end is not ready to receive them
|
||||
// as the agent will re-apply the same configuration.
|
||||
select {
|
||||
case h.cfgChan <- dynamic.Message{ProviderName: "hub", Configuration: cfg}:
|
||||
atomic.StoreInt64(&h.lastCfgUnixNano, payload.UnixNano)
|
||||
default:
|
||||
}
|
||||
}
|
||||
|
||||
func (h *handler) handleDiscoverIP(rw http.ResponseWriter, req *http.Request) {
|
||||
if req.Method != http.MethodGet {
|
||||
http.Error(rw, http.StatusText(http.StatusMethodNotAllowed), http.StatusMethodNotAllowed)
|
||||
return
|
||||
}
|
||||
|
||||
xff := req.Header.Get("X-Forwarded-For")
|
||||
port := req.URL.Query().Get("port")
|
||||
nonce := req.URL.Query().Get("nonce")
|
||||
|
||||
if err := h.doDiscoveryReq(req.Context(), xff, port, nonce); err != nil {
|
||||
err = fmt.Errorf("doing discovery request: %w", err)
|
||||
log.WithoutContext().Errorf("Handling IP discovery: %v", err)
|
||||
http.Error(rw, http.StatusText(http.StatusBadGateway), http.StatusBadGateway)
|
||||
return
|
||||
}
|
||||
|
||||
if err := json.NewEncoder(rw).Encode(xff); err != nil {
|
||||
err = fmt.Errorf("encoding discover ip response: %w", err)
|
||||
log.WithoutContext().Errorf("Handling IP discovery: %v", err)
|
||||
http.Error(rw, http.StatusText(http.StatusInternalServerError), http.StatusInternalServerError)
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
func (h *handler) doDiscoveryReq(ctx context.Context, ip, port, nonce string) error {
|
||||
req, err := http.NewRequestWithContext(ctx, http.MethodGet, fmt.Sprintf("https://%s:%s", ip, port), http.NoBody)
|
||||
if err != nil {
|
||||
return fmt.Errorf("creating request: %w", err)
|
||||
}
|
||||
|
||||
q := make(url.Values)
|
||||
q.Set("nonce", nonce)
|
||||
req.URL.RawQuery = q.Encode()
|
||||
req.Host = "agent.traefik"
|
||||
|
||||
resp, err := h.client.Do(req)
|
||||
if err != nil {
|
||||
return fmt.Errorf("doing request: %w", err)
|
||||
}
|
||||
defer func() { _ = resp.Body.Close() }()
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
type stateResponse struct {
|
||||
LastConfigUnixNano int64 `json:"lastConfigUnixNano"`
|
||||
}
|
||||
|
||||
func (h *handler) handleState(rw http.ResponseWriter, req *http.Request) {
|
||||
if req.Method != http.MethodGet {
|
||||
http.Error(rw, http.StatusText(http.StatusMethodNotAllowed), http.StatusMethodNotAllowed)
|
||||
return
|
||||
}
|
||||
|
||||
resp := stateResponse{
|
||||
LastConfigUnixNano: atomic.LoadInt64(&h.lastCfgUnixNano),
|
||||
}
|
||||
if err := json.NewEncoder(rw).Encode(resp); err != nil {
|
||||
err = fmt.Errorf("encoding last config received response: %w", err)
|
||||
log.WithoutContext().Errorf("Handling state: %v", err)
|
||||
http.Error(rw, http.StatusText(http.StatusBadRequest), http.StatusBadRequest)
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
func (h *handler) ServeHTTP(rw http.ResponseWriter, req *http.Request) {
|
||||
h.mux.ServeHTTP(rw, req)
|
||||
}
|
||||
168
pkg/provider/hub/handler_test.go
Normal file
168
pkg/provider/hub/handler_test.go
Normal file
|
|
@ -0,0 +1,168 @@
|
|||
package hub
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"crypto/tls"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"net"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"net/url"
|
||||
"strconv"
|
||||
"testing"
|
||||
"time"
|
||||
|
||||
"github.com/stretchr/testify/assert"
|
||||
"github.com/stretchr/testify/require"
|
||||
"github.com/traefik/traefik/v2/pkg/config/dynamic"
|
||||
"github.com/traefik/traefik/v2/pkg/tls/generate"
|
||||
)
|
||||
|
||||
func TestHandleConfig(t *testing.T) {
|
||||
cfgChan := make(chan dynamic.Message, 1)
|
||||
|
||||
client, err := createAgentClient(&TLS{Insecure: true})
|
||||
require.NoError(t, err)
|
||||
h := newHandler("traefik-hub-ep", 42, cfgChan, nil, client)
|
||||
|
||||
cfg := emptyDynamicConfiguration()
|
||||
cfg.HTTP.Routers["foo"] = &dynamic.Router{
|
||||
EntryPoints: []string{"ep"},
|
||||
Service: "bar",
|
||||
Rule: "Host(`foo.com`)",
|
||||
}
|
||||
|
||||
req := configRequest{Configuration: cfg}
|
||||
|
||||
b, err := json.Marshal(req)
|
||||
require.NoError(t, err)
|
||||
|
||||
server := httptest.NewServer(h)
|
||||
t.Cleanup(server.Close)
|
||||
|
||||
resp, err := http.Post(server.URL+"/config", "application/json", bytes.NewReader(b))
|
||||
require.NoError(t, err)
|
||||
require.Equal(t, http.StatusOK, resp.StatusCode)
|
||||
|
||||
select {
|
||||
case gotCfgRaw := <-cfgChan:
|
||||
patchDynamicConfiguration(cfg, "traefik-hub-ep", 42, nil)
|
||||
assert.Equal(t, cfg, gotCfgRaw.Configuration)
|
||||
|
||||
case <-time.After(time.Second):
|
||||
t.Fatal("Configuration not received")
|
||||
}
|
||||
}
|
||||
|
||||
func TestHandle_Config_MethodNotAllowed(t *testing.T) {
|
||||
cfgChan := make(chan dynamic.Message, 1)
|
||||
client, err := createAgentClient(&TLS{Insecure: true})
|
||||
require.NoError(t, err)
|
||||
h := newHandler("traefik-hub-ep", 42, cfgChan, nil, client)
|
||||
|
||||
server := httptest.NewServer(h)
|
||||
t.Cleanup(server.Close)
|
||||
|
||||
resp, err := http.Get(server.URL + "/config")
|
||||
require.NoError(t, err)
|
||||
|
||||
err = resp.Body.Close()
|
||||
require.NoError(t, err)
|
||||
|
||||
assert.Equal(t, http.StatusMethodNotAllowed, resp.StatusCode)
|
||||
}
|
||||
|
||||
func TestHandle_DiscoverIP(t *testing.T) {
|
||||
listener, err := net.Listen("tcp", "127.0.0.1:0")
|
||||
require.NoError(t, err)
|
||||
|
||||
port := listener.Addr().(*net.TCPAddr).Port
|
||||
nonce := "XVlBzgbaiCMRAjWw"
|
||||
|
||||
mux := http.NewServeMux()
|
||||
|
||||
var handlerCallCount int
|
||||
mux.HandleFunc("/", func(_ http.ResponseWriter, req *http.Request) {
|
||||
handlerCallCount++
|
||||
assert.Equal(t, nonce, req.URL.Query().Get("nonce"))
|
||||
})
|
||||
|
||||
certificate, err := generate.DefaultCertificate()
|
||||
require.NoError(t, err)
|
||||
agentServer := &http.Server{
|
||||
Handler: mux,
|
||||
TLSConfig: &tls.Config{
|
||||
Certificates: []tls.Certificate{*certificate},
|
||||
InsecureSkipVerify: true,
|
||||
MinVersion: tls.VersionTLS13,
|
||||
},
|
||||
}
|
||||
t.Cleanup(func() { _ = agentServer.Close() })
|
||||
|
||||
rdy := make(chan struct{})
|
||||
|
||||
go func(s *http.Server) {
|
||||
close(rdy)
|
||||
if err = s.ServeTLS(listener, "", ""); errors.Is(err, http.ErrServerClosed) {
|
||||
return
|
||||
}
|
||||
}(agentServer)
|
||||
|
||||
<-rdy
|
||||
|
||||
cfgChan := make(chan dynamic.Message, 1)
|
||||
client, err := createAgentClient(&TLS{Insecure: true})
|
||||
require.NoError(t, err)
|
||||
h := newHandler("traefik-hub-ep", 42, cfgChan, nil, client)
|
||||
|
||||
traefikServer := httptest.NewServer(h)
|
||||
t.Cleanup(traefikServer.Close)
|
||||
|
||||
req, err := http.NewRequest(http.MethodGet, traefikServer.URL+"/discover-ip", http.NoBody)
|
||||
require.NoError(t, err)
|
||||
|
||||
q := make(url.Values)
|
||||
q.Set("port", strconv.Itoa(port))
|
||||
q.Set("nonce", nonce)
|
||||
req.URL.RawQuery = q.Encode()
|
||||
|
||||
// Simulate a call from behind different proxies.
|
||||
req.Header.Add("X-Forwarded-For", "127.0.0.1")
|
||||
req.Header.Add("X-Forwarded-For", "10.10.0.13")
|
||||
|
||||
resp, err := http.DefaultClient.Do(req)
|
||||
require.NoError(t, err)
|
||||
|
||||
defer func() {
|
||||
err = resp.Body.Close()
|
||||
require.NoError(t, err)
|
||||
}()
|
||||
|
||||
assert.Equal(t, 1, handlerCallCount)
|
||||
assert.Equal(t, http.StatusOK, resp.StatusCode)
|
||||
|
||||
var ip string
|
||||
err = json.NewDecoder(resp.Body).Decode(&ip)
|
||||
require.NoError(t, err)
|
||||
|
||||
assert.Equal(t, "127.0.0.1", ip)
|
||||
}
|
||||
|
||||
func TestHandle_DiscoverIP_MethodNotAllowed(t *testing.T) {
|
||||
cfgChan := make(chan dynamic.Message, 1)
|
||||
client, err := createAgentClient(&TLS{Insecure: true})
|
||||
require.NoError(t, err)
|
||||
h := newHandler("traefik-hub-ep", 42, cfgChan, nil, client)
|
||||
|
||||
server := httptest.NewServer(h)
|
||||
t.Cleanup(server.Close)
|
||||
|
||||
resp, err := http.Post(server.URL+"/discover-ip", "", http.NoBody)
|
||||
require.NoError(t, err)
|
||||
|
||||
err = resp.Body.Close()
|
||||
require.NoError(t, err)
|
||||
|
||||
assert.Equal(t, http.StatusMethodNotAllowed, resp.StatusCode)
|
||||
}
|
||||
215
pkg/provider/hub/hub.go
Normal file
215
pkg/provider/hub/hub.go
Normal file
|
|
@ -0,0 +1,215 @@
|
|||
package hub
|
||||
|
||||
import (
|
||||
"crypto/tls"
|
||||
"crypto/x509"
|
||||
"errors"
|
||||
"fmt"
|
||||
"net"
|
||||
"net/http"
|
||||
|
||||
"github.com/traefik/traefik/v2/pkg/config/dynamic"
|
||||
"github.com/traefik/traefik/v2/pkg/log"
|
||||
"github.com/traefik/traefik/v2/pkg/provider"
|
||||
"github.com/traefik/traefik/v2/pkg/safe"
|
||||
ttls "github.com/traefik/traefik/v2/pkg/tls"
|
||||
)
|
||||
|
||||
var _ provider.Provider = (*Provider)(nil)
|
||||
|
||||
// DefaultEntryPointName is the name of the default internal entry point.
|
||||
const DefaultEntryPointName = "traefik-hub"
|
||||
|
||||
// Provider holds configurations of the provider.
|
||||
type Provider struct {
|
||||
EntryPoint string `description:"Entrypoint that exposes data for Traefik Hub. It should be a dedicated one, and not used by any router." json:"entryPoint,omitempty" toml:"entryPoint,omitempty" yaml:"entryPoint,omitempty" export:"true"`
|
||||
TLS *TLS `description:"TLS configuration for mTLS communication between Traefik and Hub Agent." json:"tls,omitempty" toml:"tls,omitempty" yaml:"tls,omitempty" export:"true"`
|
||||
|
||||
server *http.Server
|
||||
}
|
||||
|
||||
// TLS configures the mTLS connection between Traefik Proxy and the Traefik Hub Agent.
|
||||
type TLS struct {
|
||||
Insecure bool `description:"Enables an insecure TLS connection that uses default credentials, and which has no peer authentication between Traefik Proxy and the Traefik Hub Agent." json:"insecure,omitempty" toml:"insecure,omitempty" yaml:"insecure,omitempty" export:"true"`
|
||||
CA ttls.FileOrContent `description:"The certificate authority authenticates the Traefik Hub Agent certificate." json:"ca,omitempty" toml:"ca,omitempty" yaml:"ca,omitempty" loggable:"false"`
|
||||
Cert ttls.FileOrContent `description:"The TLS certificate for Traefik Proxy as a TLS client." json:"cert,omitempty" toml:"cert,omitempty" yaml:"cert,omitempty" loggable:"false"`
|
||||
Key ttls.FileOrContent `description:"The TLS key for Traefik Proxy as a TLS client." json:"key,omitempty" toml:"key,omitempty" yaml:"key,omitempty" loggable:"false"`
|
||||
}
|
||||
|
||||
// SetDefaults sets the default values.
|
||||
func (p *Provider) SetDefaults() {
|
||||
p.EntryPoint = DefaultEntryPointName
|
||||
}
|
||||
|
||||
// Init the provider.
|
||||
func (p *Provider) Init() error {
|
||||
return nil
|
||||
}
|
||||
|
||||
// Provide allows the hub provider to provide configurations to traefik using the given configuration channel.
|
||||
func (p *Provider) Provide(configurationChan chan<- dynamic.Message, _ *safe.Pool) error {
|
||||
listener, err := net.Listen("tcp", "127.0.0.1:0")
|
||||
if err != nil {
|
||||
return fmt.Errorf("listener: %w", err)
|
||||
}
|
||||
port := listener.Addr().(*net.TCPAddr).Port
|
||||
|
||||
client, err := createAgentClient(p.TLS)
|
||||
if err != nil {
|
||||
return fmt.Errorf("creating Hub Agent HTTP client: %w", err)
|
||||
}
|
||||
|
||||
p.server = &http.Server{Handler: newHandler(p.EntryPoint, port, configurationChan, p.TLS, client)}
|
||||
|
||||
// TODO: this is going to be leaky (because no context to make it terminate)
|
||||
// if/when Provide lifecycle differs with Traefik lifecycle.
|
||||
go func() {
|
||||
if err = p.server.Serve(listener); err != nil {
|
||||
log.WithoutContext().WithField(log.ProviderName, "hub").Errorf("Unexpected error while running server: %v", err)
|
||||
return
|
||||
}
|
||||
}()
|
||||
|
||||
exposeAPIAndMetrics(configurationChan, p.EntryPoint, port, p.TLS)
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func exposeAPIAndMetrics(cfgChan chan<- dynamic.Message, ep string, port int, tlsCfg *TLS) {
|
||||
cfg := emptyDynamicConfiguration()
|
||||
|
||||
patchDynamicConfiguration(cfg, ep, port, tlsCfg)
|
||||
|
||||
cfgChan <- dynamic.Message{ProviderName: "hub", Configuration: cfg}
|
||||
}
|
||||
|
||||
func patchDynamicConfiguration(cfg *dynamic.Configuration, ep string, port int, tlsCfg *TLS) {
|
||||
cfg.HTTP.Routers["traefik-hub-agent-api"] = &dynamic.Router{
|
||||
EntryPoints: []string{ep},
|
||||
Service: "api@internal",
|
||||
Rule: "Host(`proxy.traefik`) && PathPrefix(`/api`)",
|
||||
}
|
||||
cfg.HTTP.Routers["traefik-hub-agent-metrics"] = &dynamic.Router{
|
||||
EntryPoints: []string{ep},
|
||||
Service: "prometheus@internal",
|
||||
Rule: "Host(`proxy.traefik`) && PathPrefix(`/metrics`)",
|
||||
}
|
||||
|
||||
cfg.HTTP.Routers["traefik-hub-agent-service"] = &dynamic.Router{
|
||||
EntryPoints: []string{ep},
|
||||
Service: "traefik-hub-agent-service",
|
||||
Rule: "Host(`proxy.traefik`) && PathPrefix(`/config`, `/discover-ip`, `/state`)",
|
||||
}
|
||||
|
||||
cfg.HTTP.Services["traefik-hub-agent-service"] = &dynamic.Service{
|
||||
LoadBalancer: &dynamic.ServersLoadBalancer{
|
||||
Servers: []dynamic.Server{
|
||||
{
|
||||
URL: fmt.Sprintf("http://127.0.0.1:%d", port),
|
||||
},
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
if tlsCfg == nil {
|
||||
return
|
||||
}
|
||||
|
||||
if tlsCfg.Insecure {
|
||||
cfg.TLS.Options["traefik-hub"] = ttls.Options{
|
||||
MinVersion: "VersionTLS13",
|
||||
}
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
cfg.TLS.Options["traefik-hub"] = ttls.Options{
|
||||
ClientAuth: ttls.ClientAuth{
|
||||
CAFiles: []ttls.FileOrContent{tlsCfg.CA},
|
||||
ClientAuthType: "RequireAndVerifyClientCert",
|
||||
},
|
||||
SniStrict: true,
|
||||
MinVersion: "VersionTLS13",
|
||||
}
|
||||
|
||||
cfg.TLS.Certificates = append(cfg.TLS.Certificates, &ttls.CertAndStores{
|
||||
Certificate: ttls.Certificate{
|
||||
CertFile: tlsCfg.Cert,
|
||||
KeyFile: tlsCfg.Key,
|
||||
},
|
||||
})
|
||||
}
|
||||
|
||||
func emptyDynamicConfiguration() *dynamic.Configuration {
|
||||
return &dynamic.Configuration{
|
||||
HTTP: &dynamic.HTTPConfiguration{
|
||||
Routers: make(map[string]*dynamic.Router),
|
||||
Middlewares: make(map[string]*dynamic.Middleware),
|
||||
Services: make(map[string]*dynamic.Service),
|
||||
ServersTransports: make(map[string]*dynamic.ServersTransport),
|
||||
},
|
||||
TCP: &dynamic.TCPConfiguration{
|
||||
Routers: make(map[string]*dynamic.TCPRouter),
|
||||
Services: make(map[string]*dynamic.TCPService),
|
||||
},
|
||||
TLS: &dynamic.TLSConfiguration{
|
||||
Stores: make(map[string]ttls.Store),
|
||||
Options: make(map[string]ttls.Options),
|
||||
},
|
||||
UDP: &dynamic.UDPConfiguration{
|
||||
Routers: make(map[string]*dynamic.UDPRouter),
|
||||
Services: make(map[string]*dynamic.UDPService),
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
func createAgentClient(tlsCfg *TLS) (http.Client, error) {
|
||||
var client http.Client
|
||||
if tlsCfg.Insecure {
|
||||
client.Transport = &http.Transport{
|
||||
TLSClientConfig: &tls.Config{
|
||||
InsecureSkipVerify: true,
|
||||
MinVersion: tls.VersionTLS13,
|
||||
},
|
||||
}
|
||||
|
||||
return client, nil
|
||||
}
|
||||
|
||||
caContent, err := tlsCfg.CA.Read()
|
||||
if err != nil {
|
||||
return client, fmt.Errorf("reading CA: %w", err)
|
||||
}
|
||||
|
||||
roots := x509.NewCertPool()
|
||||
if ok := roots.AppendCertsFromPEM(caContent); !ok {
|
||||
return client, errors.New("appending CA error")
|
||||
}
|
||||
|
||||
certContent, err := tlsCfg.Cert.Read()
|
||||
if err != nil {
|
||||
return client, fmt.Errorf("reading Cert: %w", err)
|
||||
}
|
||||
keyContent, err := tlsCfg.Key.Read()
|
||||
if err != nil {
|
||||
return client, fmt.Errorf("reading Key: %w", err)
|
||||
}
|
||||
|
||||
certificate, err := tls.X509KeyPair(certContent, keyContent)
|
||||
if err != nil {
|
||||
return client, fmt.Errorf("creating key pair: %w", err)
|
||||
}
|
||||
|
||||
// mTLS
|
||||
client.Transport = &http.Transport{
|
||||
TLSClientConfig: &tls.Config{
|
||||
RootCAs: roots,
|
||||
Certificates: []tls.Certificate{certificate},
|
||||
ServerName: "agent.traefik",
|
||||
ClientAuth: tls.RequireAndVerifyClientCert,
|
||||
MinVersion: tls.VersionTLS13,
|
||||
},
|
||||
}
|
||||
|
||||
return client, nil
|
||||
}
|
||||
Loading…
Add table
Add a link
Reference in a new issue