traefik/pkg/pilot/pilot.go

package pilot

import (
	"bytes"
	"context"
	"encoding/json"
	"fmt"
	"io/ioutil"
	"net/http"
	"time"

	"github.com/cenkalti/backoff/v4"
	"github.com/traefik/traefik/v2/pkg/config/runtime"
	"github.com/traefik/traefik/v2/pkg/log"
	"github.com/traefik/traefik/v2/pkg/metrics"
	"github.com/traefik/traefik/v2/pkg/safe"
	"github.com/traefik/traefik/v2/pkg/version"
)

const baseURL = "https://instance-info.pilot.traefik.io/public"

const tokenHeader = "X-Token"

const (
	pilotTimer     = 5 * time.Minute
	maxElapsedTime = 4 * time.Minute
)

// RunTimeRepresentation is the configuration information exposed by the API handler.
type RunTimeRepresentation struct {
	Routers     map[string]*runtime.RouterInfo        `json:"routers,omitempty"`
	Middlewares map[string]*runtime.MiddlewareInfo    `json:"middlewares,omitempty"`
	Services    map[string]*serviceInfoRepresentation `json:"services,omitempty"`
	TCPRouters  map[string]*runtime.TCPRouterInfo     `json:"tcpRouters,omitempty"`
	TCPServices map[string]*runtime.TCPServiceInfo    `json:"tcpServices,omitempty"`
	UDPRouters  map[string]*runtime.UDPRouterInfo     `json:"udpRouters,omitempty"`
	UDPServices map[string]*runtime.UDPServiceInfo    `json:"udpServices,omitempty"`
}

type serviceInfoRepresentation struct {
	*runtime.ServiceInfo
	ServerStatus map[string]string `json:"serverStatus,omitempty"`
}

type instanceInfo struct {
	ID            string                `json:"id,omitempty"`
	Configuration RunTimeRepresentation `json:"configuration,omitempty"`
	Metrics       []metrics.PilotMetric `json:"metrics,omitempty"`
}

// New creates a new Pilot.
func New(token string, metricsRegistry *metrics.PilotRegistry, pool *safe.Pool) *Pilot {
	return &Pilot{
		rtConfChan: make(chan *runtime.Configuration),
		client: &client{
			token:      token,
			httpClient: &http.Client{Timeout: 5 * time.Second},
			baseURL:    baseURL,
		},
		routinesPool:    pool,
		metricsRegistry: metricsRegistry,
	}
}

// Pilot connector with Pilot.
type Pilot struct {
	routinesPool *safe.Pool
	client       *client

	rtConf          *runtime.Configuration
	rtConfChan      chan *runtime.Configuration
	metricsRegistry *metrics.PilotRegistry
}

// SetRuntimeConfiguration stores the runtime configuration.
func (p *Pilot) SetRuntimeConfiguration(rtConf *runtime.Configuration) {
	p.rtConfChan <- rtConf
}

func (p *Pilot) getRepresentation() RunTimeRepresentation {
	if p.rtConf == nil {
		return RunTimeRepresentation{}
	}

	siRepr := make(map[string]*serviceInfoRepresentation, len(p.rtConf.Services))
	for k, v := range p.rtConf.Services {
		siRepr[k] = &serviceInfoRepresentation{
			ServiceInfo:  v,
			ServerStatus: v.GetAllStatus(),
		}
	}

	result := RunTimeRepresentation{
		Routers:     p.rtConf.Routers,
		Middlewares: p.rtConf.Middlewares,
		Services:    siRepr,
		TCPRouters:  p.rtConf.TCPRouters,
		TCPServices: p.rtConf.TCPServices,
		UDPRouters:  p.rtConf.UDPRouters,
		UDPServices: p.rtConf.UDPServices,
	}

	return result
}

func (p *Pilot) sendData(ctx context.Context, conf RunTimeRepresentation, pilotMetrics []metrics.PilotMetric) {
	err := p.client.SendData(ctx, conf, pilotMetrics)
	if err != nil {
		log.WithoutContext().Error(err)
	}
}

// Tick sends data periodically.
func (p *Pilot) Tick(ctx context.Context) {
	select {
	case rtConf := <-p.rtConfChan:
		p.rtConf = rtConf
		break
	case <-ctx.Done():
		return
	}

	conf := p.getRepresentation()
	pilotMetrics := p.metricsRegistry.Data()

	p.routinesPool.GoCtx(func(ctxRt context.Context) {
		p.sendData(ctxRt, conf, pilotMetrics)
	})

	ticker := time.NewTicker(pilotTimer)
	for {
		select {
		case tick := <-ticker.C:
			log.WithoutContext().Debugf("Send to pilot: %s", tick)

			conf := p.getRepresentation()
			pilotMetrics := p.metricsRegistry.Data()

			p.routinesPool.GoCtx(func(ctxRt context.Context) {
				p.sendData(ctxRt, conf, pilotMetrics)
			})
		case rtConf := <-p.rtConfChan:
			p.rtConf = rtConf
		case <-ctx.Done():
			return
		}
	}
}

type client struct {
	httpClient *http.Client
	baseURL    string
	token      string
	uuid       string
}

func (c *client) createUUID() (string, error) {
	data := []byte(`{"version":"` + version.Version + `","codeName":"` + version.Codename + `"}`)
	req, err := http.NewRequest(http.MethodPost, c.baseURL+"/", bytes.NewBuffer(data))
	if err != nil {
		return "", fmt.Errorf("failed to create request: %w", err)
	}

	req.Header.Set("Content-Type", "application/json")
	req.Header.Set(tokenHeader, c.token)

	resp, err := c.httpClient.Do(req)
	if err != nil {
		return "", fmt.Errorf("failed call Pilot: %w", err)
	}

	defer resp.Body.Close()

	body, err := ioutil.ReadAll(resp.Body)
	if err != nil {
		return "", fmt.Errorf("failed read response body: %w", err)
	}

	if resp.StatusCode/100 != 2 {
		return "", fmt.Errorf("wrong status code while sending configuration: %d: %s", resp.StatusCode, body)
	}

	created := instanceInfo{}
	err = json.Unmarshal(body, &created)
	if err != nil {
		return "", fmt.Errorf("failed to unmarshal response body: %w", err)
	}

	return created.ID, nil
}

// SendData sends data to Pilot.
func (c *client) SendData(ctx context.Context, rtConf RunTimeRepresentation, pilotMetrics []metrics.PilotMetric) error {
	exponentialBackOff := backoff.NewExponentialBackOff()
	exponentialBackOff.MaxElapsedTime = maxElapsedTime

	return backoff.RetryNotify(
		func() error {
			return c.sendData(rtConf, pilotMetrics)
		},
		backoff.WithContext(exponentialBackOff, ctx),
		func(err error, duration time.Duration) {
			log.WithoutContext().Errorf("retry in %s due to: %v ", duration, err)
		})
}

func (c *client) sendData(_ RunTimeRepresentation, pilotMetrics []metrics.PilotMetric) error {
	if len(c.uuid) == 0 {
		var err error
		c.uuid, err = c.createUUID()
		if err != nil {
			return fmt.Errorf("failed to create UUID: %w", err)
		}

		version.UUID = c.uuid
	}

	info := instanceInfo{
		ID:      c.uuid,
		Metrics: pilotMetrics,
	}

	b, err := json.Marshal(info)
	if err != nil {
		return fmt.Errorf("failed to marshall request body: %w", err)
	}

	request, err := http.NewRequest(http.MethodPost, c.baseURL+"/command", bytes.NewBuffer(b))
	if err != nil {
		return fmt.Errorf("failed to create request: %w", err)
	}

	request.Header.Set("Content-Type", "application/json")
	request.Header.Set(tokenHeader, c.token)

	resp, err := c.httpClient.Do(request)
	if err != nil {
		return fmt.Errorf("failed to call Pilot: %w", err)
	}

	defer resp.Body.Close()

	body, err := ioutil.ReadAll(resp.Body)
	if err != nil {
		return fmt.Errorf("failed to read response body: %w", err)
	}

	if resp.StatusCode != http.StatusOK {
		return fmt.Errorf("wrong status code while sending configuration: %d: %s", resp.StatusCode, body)
	}

	return nil
}