prometheus/discovery/xds/client.go

// Copyright 2021 The Prometheus Authors
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

package xds

import (
	"bytes"
	"context"
	"errors"
	"fmt"
	"io"
	"net/http"
	"net/url"
	"path"
	"time"

	envoy_core "github.com/envoyproxy/go-control-plane/envoy/config/core/v3"
	v3 "github.com/envoyproxy/go-control-plane/envoy/service/discovery/v3"
	"github.com/prometheus/common/config"
	"github.com/prometheus/common/version"
)

var userAgent = fmt.Sprintf("Prometheus/%s", version.Version)

// ResourceClient exposes the xDS protocol for a single resource type.
// See https://www.envoyproxy.io/docs/envoy/latest/api-docs/xds_protocol#rest-json-polling-subscriptions .
type ResourceClient interface {
	// ResourceTypeURL is the type URL of the resource.
	ResourceTypeURL() string

	// Server is the xDS Management server.
	Server() string

	// Fetch requests the latest view of the entire resource state.
	// If no updates have been made since the last request, the response will be nil.
	Fetch(ctx context.Context) (*v3.DiscoveryResponse, error)

	// ID returns the ID of the client that is sent to the xDS server.
	ID() string

	// Close releases currently held resources.
	Close()
}

type HTTPResourceClient struct {
	client *http.Client
	config *HTTPResourceClientConfig

	// endpoint is the fully-constructed xDS HTTP endpoint.
	endpoint string
	// Caching.
	latestVersion string
	latestNonce   string
}

type HTTPResourceClientConfig struct {
	// HTTP config.
	config.HTTPClientConfig

	Name string

	// ExtraQueryParams are extra query parameters to attach to the request URL.
	ExtraQueryParams url.Values

	// General xDS config.

	// The timeout for a single fetch request.
	Timeout time.Duration

	// Type is the xds type, e.g., clusters
	// which is used in the discovery POST request.
	ResourceType string
	// ResourceTypeURL is the Google type url for the resource, e.g., type.googleapis.com/envoy.api.v2.Cluster.
	ResourceTypeURL string
	// Server is the xDS management server.
	Server string
	// ClientID is used to identify the client with the management server.
	ClientID string
}

func NewHTTPResourceClient(conf *HTTPResourceClientConfig, protocolVersion ProtocolVersion) (*HTTPResourceClient, error) {
	if protocolVersion != ProtocolV3 {
		return nil, errors.New("only the v3 protocol is supported")
	}

	if len(conf.Server) == 0 {
		return nil, errors.New("empty xDS server")
	}

	serverURL, err := url.Parse(conf.Server)
	if err != nil {
		return nil, err
	}

	endpointURL, err := makeXDSResourceHTTPEndpointURL(protocolVersion, serverURL, conf.ResourceType)
	if err != nil {
		return nil, err
	}

	if conf.ExtraQueryParams != nil {
		endpointURL.RawQuery = conf.ExtraQueryParams.Encode()
	}

	client, err := config.NewClientFromConfig(conf.HTTPClientConfig, conf.Name, config.WithIdleConnTimeout(conf.Timeout))
	if err != nil {
		return nil, err
	}

	client.Timeout = conf.Timeout

	return &HTTPResourceClient{
		client:        client,
		config:        conf,
		endpoint:      endpointURL.String(),
		latestVersion: "",
		latestNonce:   "",
	}, nil
}

func makeXDSResourceHTTPEndpointURL(protocolVersion ProtocolVersion, serverURL *url.URL, resourceType string) (*url.URL, error) {
	if serverURL == nil {
		return nil, errors.New("empty xDS server URL")
	}

	if len(serverURL.Scheme) == 0 || len(serverURL.Host) == 0 {
		return nil, errors.New("invalid xDS server URL")
	}

	if serverURL.Scheme != "http" && serverURL.Scheme != "https" {
		return nil, errors.New("invalid xDS server URL protocol. must be either 'http' or 'https'")
	}

	serverURL.Path = path.Join(serverURL.Path, string(protocolVersion), fmt.Sprintf("discovery:%s", resourceType))

	return serverURL, nil
}

func (rc *HTTPResourceClient) Server() string {
	return rc.config.Server
}

func (rc *HTTPResourceClient) ResourceTypeURL() string {
	return rc.config.ResourceTypeURL
}

func (rc *HTTPResourceClient) ID() string {
	return rc.config.ClientID
}

func (rc *HTTPResourceClient) Close() {
	rc.client.CloseIdleConnections()
}

// Fetch requests the latest state of the resources from the xDS server and cache the version.
// Returns a nil response if the current local version is up to date.
func (rc *HTTPResourceClient) Fetch(ctx context.Context) (*v3.DiscoveryResponse, error) {
	discoveryReq := &v3.DiscoveryRequest{
		VersionInfo:   rc.latestVersion,
		ResponseNonce: rc.latestNonce,
		TypeUrl:       rc.ResourceTypeURL(),
		ResourceNames: []string{},
		Node: &envoy_core.Node{
			Id: rc.ID(),
		},
	}

	reqBody, err := protoJSONMarshalOptions.Marshal(discoveryReq)
	if err != nil {
		return nil, err
	}

	request, err := http.NewRequest("POST", rc.endpoint, bytes.NewBuffer(reqBody))
	if err != nil {
		return nil, err
	}
	request = request.WithContext(ctx)

	request.Header.Add("User-Agent", userAgent)
	request.Header.Add("Content-Type", "application/json")
	request.Header.Add("Accept", "application/json")

	resp, err := rc.client.Do(request)
	if err != nil {
		return nil, err
	}
	defer func() {
		io.Copy(io.Discard, resp.Body)
		resp.Body.Close()
	}()

	if resp.StatusCode == http.StatusNotModified {
		// Empty response, already have the latest.
		return nil, nil
	}

	if resp.StatusCode != http.StatusOK {
		return nil, fmt.Errorf("non 200 status '%d' response during xDS fetch", resp.StatusCode)
	}

	respBody, err := io.ReadAll(resp.Body)
	if err != nil {
		return nil, err
	}

	discoveryRes := &v3.DiscoveryResponse{}
	if err = protoJSONUnmarshalOptions.Unmarshal(respBody, discoveryRes); err != nil {
		return nil, err
	}

	// Cache the latest nonce + version info.
	rc.latestNonce = discoveryRes.Nonce
	rc.latestVersion = discoveryRes.VersionInfo

	return discoveryRes, nil
}