VictoriaMetrics/lib/promscrape/discovery/yandexcloud/api.go

package yandexcloud

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

	"github.com/VictoriaMetrics/VictoriaMetrics/lib/logger"
	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promauth"
	"github.com/VictoriaMetrics/VictoriaMetrics/lib/promscrape/discoveryutils"
)

var configMap = discoveryutils.NewConfigMap()

type apiCredentials struct {
	Token      string    `json:"Token"`
	Expiration time.Time `json:"Expiration"`
}

// yandexPassportOAuth is a struct for Yandex Cloud IAM token request
// https://cloud.yandex.com/en-ru/docs/iam/operations/iam-token/create
type yandexPassportOAuth struct {
	YandexPassportOAuthToken string `json:"yandexPassportOauthToken"`
}

type apiConfig struct {
	client              *http.Client
	yandexPassportOAuth *yandexPassportOAuth
	serviceEndpoints    map[string]string

	// credsLock protects the refresh of creds
	credsLock sync.Mutex
	creds     *apiCredentials
}

func getAPIConfig(sdc *SDConfig, baseDir string) (*apiConfig, error) {
	v, err := configMap.Get(sdc, func() (interface{}, error) { return newAPIConfig(sdc, baseDir) })
	if err != nil {
		return nil, err
	}
	return v.(*apiConfig), nil
}

func newAPIConfig(sdc *SDConfig, baseDir string) (*apiConfig, error) {
	transport := &http.Transport{
		MaxIdleConnsPerHost: 100,
	}
	if sdc.TLSConfig != nil {
		opts := &promauth.Options{
			BaseDir:   baseDir,
			TLSConfig: sdc.TLSConfig,
		}
		ac, err := opts.NewConfig()
		if err != nil {
			return nil, fmt.Errorf("cannot initialize TLS config: %w", err)
		}
		transport.TLSClientConfig = ac.NewTLSConfig()
	}
	cfg := &apiConfig{
		client: &http.Client{
			Transport: transport,
		},
	}
	apiEndpoint := sdc.APIEndpoint
	if apiEndpoint == "" {
		apiEndpoint = "https://api.cloud.yandex.net"
	}
	serviceEndpoints, err := cfg.getServiceEndpoints(apiEndpoint)
	if err != nil {
		return nil, fmt.Errorf("cannot obtain endpoints for yandex services: %w", err)
	}
	cfg.serviceEndpoints = serviceEndpoints
	if sdc.YandexPassportOAuthToken != nil {
		logger.Infof("yandexcloud_sd: using yandex passport OAuth token")
		cfg.yandexPassportOAuth = &yandexPassportOAuth{
			YandexPassportOAuthToken: sdc.YandexPassportOAuthToken.String(),
		}
	}
	return cfg, nil
}

// getFreshAPICredentials checks token lifetime and update if needed
func (cfg *apiConfig) getFreshAPICredentials() (*apiCredentials, error) {
	cfg.credsLock.Lock()
	defer cfg.credsLock.Unlock()

	if cfg.creds != nil && time.Until(cfg.creds.Expiration) > 10*time.Second {
		// Credentials aren't expired yet.
		return cfg.creds, nil
	}
	// Refresh credentials.
	newCreds, err := getCreds(cfg)
	if err != nil {
		return nil, fmt.Errorf("cannot refresh service account api token: %w", err)
	}
	cfg.creds = newCreds
	logger.Infof("yandexcloud_sd: successfully refreshed service account api token; expiration: %.3f seconds", time.Until(newCreds.Expiration).Seconds())
	return newCreds, nil
}

// getCreds get Yandex Cloud IAM token based on configuration
func getCreds(cfg *apiConfig) (*apiCredentials, error) {
	if cfg.yandexPassportOAuth == nil {
		return getInstanceCreds(cfg)
	}
	it, err := getIAMToken(cfg)
	if err != nil {
		return nil, fmt.Errorf("cannot get IAM token: %w", err)
	}
	return &apiCredentials{
		Token:      it.IAMToken,
		Expiration: it.ExpiresAt,
	}, nil
}

// getInstanceCreds gets Yandex Cloud IAM token using instance Service Account
//
// See https://cloud.yandex.com/en-ru/docs/compute/operations/vm-connect/auth-inside-vm
func getInstanceCreds(cfg *apiConfig) (*apiCredentials, error) {
	endpoint := "http://169.254.169.254/latest/meta-data/iam/security-credentials/default"
	resp, err := cfg.client.Get(endpoint)
	if err != nil {
		return nil, fmt.Errorf("cannot read instance creds from %s: %w", endpoint, err)
	}
	data, err := readResponseBody(resp, endpoint)
	if err != nil {
		return nil, err
	}
	var ac apiCredentials
	if err := json.Unmarshal(data, &ac); err != nil {
		return nil, fmt.Errorf("cannot parse auth credentials response from %s: %w", endpoint, err)
	}
	return &ac, nil
}

// getIAMToken gets Yandex Cloud IAM token using OAuth
//
// See https://cloud.yandex.com/en-ru/docs/iam/operations/iam-token/create
func getIAMToken(cfg *apiConfig) (*iamToken, error) {
	iamURL := cfg.serviceEndpoints["iam"] + "/iam/v1/tokens"
	passport, err := json.Marshal(cfg.yandexPassportOAuth)
	if err != nil {
		logger.Panicf("BUG: cannot marshal yandex passport OAuth token: %s", err)
	}
	body := bytes.NewBuffer(passport)
	resp, err := cfg.client.Post(iamURL, "application/json", body)
	if err != nil {
		logger.Panicf("BUG: cannot create request to yandex cloud iam api %q: %s", iamURL, err)
	}
	data, err := readResponseBody(resp, iamURL)
	if err != nil {
		return nil, err
	}
	var it iamToken
	if err := json.Unmarshal(data, &it); err != nil {
		return nil, fmt.Errorf("cannot parse iam token: %w; data: %s", err, data)
	}
	return &it, nil
}

// iamToken represents Yandex Cloud IAM token response
//
// See https://cloud.yandex.com/en-ru/docs/iam/operations/iam-token/create
type iamToken struct {
	IAMToken  string    `json:"iamToken"`
	ExpiresAt time.Time `json:"expiresAt"`
}

// getServiceEndpoints returns services endpoints map
//
// See https://cloud.yandex.com/en-ru/docs/api-design-guide/concepts/endpoints
func (cfg *apiConfig) getServiceEndpoints(apiEndpoint string) (map[string]string, error) {
	apiEndpointURL, err := url.Parse(apiEndpoint)
	if err != nil {
		return nil, fmt.Errorf("cannot parse api_endpoint %q: %w", apiEndpoint, err)
	}
	scheme := apiEndpointURL.Scheme
	if scheme == "" {
		return nil, fmt.Errorf("missing scheme in api_endpoint %q", apiEndpoint)
	}
	if apiEndpointURL.Host == "" {
		return nil, fmt.Errorf("missing host in api_endpoint %q", apiEndpoint)
	}
	endpointsURL := apiEndpoint + "/endpoints"
	resp, err := cfg.client.Get(endpointsURL)
	if err != nil {
		return nil, fmt.Errorf("cannot query %q: %w", endpointsURL, err)
	}
	data, err := readResponseBody(resp, endpointsURL)
	if err != nil {
		return nil, err
	}
	var eps endpoints
	if err := json.Unmarshal(data, &eps); err != nil {
		return nil, fmt.Errorf("cannot parse API endpoints list: %w; data=%s", err, data)
	}
	m := make(map[string]string, len(eps.Endpoints))
	for _, endpoint := range eps.Endpoints {
		m[endpoint.ID] = scheme + "://" + endpoint.Address
	}
	return m, nil
}

type endpoints struct {
	Endpoints []endpoint `json:"endpoints"`
}

type endpoint struct {
	ID      string `json:"id"`
	Address string `json:"address"`
}

// getAPIResponse calls Yandex Cloud apiURL and returns response body.
func getAPIResponse(apiURL string, cfg *apiConfig) ([]byte, error) {
	creds, err := cfg.getFreshAPICredentials()
	if err != nil {
		return nil, err
	}
	req, err := http.NewRequest("GET", apiURL, nil)
	if err != nil {
		logger.Panicf("BUG: cannot create new request for yandex cloud api url %s: %s", apiURL, err)
	}
	req.Header.Set("Authorization", "Bearer "+creds.Token)
	resp, err := cfg.client.Do(req)
	if err != nil {
		return nil, fmt.Errorf("cannot query yandex cloud api url %s: %w", apiURL, err)
	}
	return readResponseBody(resp, apiURL)
}

// readResponseBody reads body from http.Response.
func readResponseBody(resp *http.Response, apiURL string) ([]byte, error) {
	data, err := ioutil.ReadAll(resp.Body)
	_ = resp.Body.Close()
	if err != nil {
		return nil, fmt.Errorf("cannot read response from %q: %w", apiURL, err)
	}
	if resp.StatusCode != http.StatusOK {
		return nil, fmt.Errorf("unexpected status code for %q; got %d; want %d; response body: %s",
			apiURL, resp.StatusCode, http.StatusOK, data)
	}
	return data, nil
}