// 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 }