feature: support metadata v2

Signed-off-by: bingshen.wbs <[email protected]>

Author: BSWANG

go.mod

@@ -108,7 +108,7 @@ require (
 	go.uber.org/zap v1.26.0 // indirect
 	golang.org/x/exp v0.0.0-20230515195305-f3d0a9c9a5cc // indirect
 	golang.org/x/oauth2 v0.21.0 // indirect
-	golang.org/x/sync v0.7.0 // indirect
+	golang.org/x/sync v0.7.0
 	golang.org/x/sys v0.21.0 // indirect
 	golang.org/x/term v0.21.0 // indirect
 	golang.org/x/text v0.16.0 // indirect

internal/config/config.go

@@ -3,11 +3,9 @@ package config
 import (
 	"encoding/json"
 	"fmt"
-	"io"
-	"net/http"
 	"os"
-	"strings"
 
+	"github.com/AliyunContainerService/alibabacloud-erdma-controller/internal/utils"
 	"k8s.io/utils/ptr"
 
 	"github.com/AliyunContainerService/alibabacloud-erdma-controller/internal/types"
@@ -29,27 +27,7 @@ var (
 
 func getRegion() (string, error) {
 	url := regionIDAddr
-	resp, err := http.Get(url)
-	if err != nil {
-		return "", fmt.Errorf("error get url: %s from metaserver. %w", url, err)
-	}
-	//nolint:errcheck
-	defer resp.Body.Close()
-
-	if resp.StatusCode == http.StatusNotFound {
-		return "", fmt.Errorf("error get url: %s from metaserver, code: %v, %v", url, resp.StatusCode, "Not Found")
-	}
-	if resp.StatusCode >= http.StatusBadRequest {
-		return "", fmt.Errorf("error get url: %s from metaserver, code: %v", url, resp.StatusCode)
-	}
-
-	body, err := io.ReadAll(resp.Body)
-	if err != nil {
-		return "", err
-	}
-	result := strings.Split(string(body), "\n")
-	trimResult := strings.Trim(result[0], "/")
-	return trimResult, nil
+	return utils.GetStrFromMetadata(url)
 }
 
 func InitConfig(configPath, credentialPath string) error {

internal/drivers/netdev.go

@@ -4,12 +4,10 @@ package drivers
 
 import (
 	"fmt"
-	"io"
 	"net"
-	"net/http"
-	"strings"
 
 	"github.com/AliyunContainerService/alibabacloud-erdma-controller/internal/types"
+	"github.com/AliyunContainerService/alibabacloud-erdma-controller/internal/utils"
 	"github.com/samber/lo"
 	"github.com/vishvananda/netlink"
 )
@@ -237,15 +235,15 @@ func genRoutesForAddr(gateway net.IP, cidr *net.IPNet) ([]*route, error) {
 }
 
 func getNetConfFromMetadata(mac string) (*netConf, error) {
-	addr, err := getStrFromMetadata(fmt.Sprintf(ipUrl, mac))
+	addr, err := utils.GetStrFromMetadata(fmt.Sprintf(ipUrl, mac))
 	if err != nil {
 		return nil, err
 	}
 	ip := net.ParseIP(addr)
 	if ip == nil {
 		return nil, fmt.Errorf("invalid ip address: %s", addr)
 	}
-	cidr, err := getStrFromMetadata(fmt.Sprintf(cidrURL, mac))
+	cidr, err := utils.GetStrFromMetadata(fmt.Sprintf(cidrURL, mac))
 	if err != nil {
 		return nil, err
 	}
@@ -254,7 +252,7 @@ func getNetConfFromMetadata(mac string) (*netConf, error) {
 		return nil, fmt.Errorf("invalid cidr: %s", cidr)
 	}
 
-	gw, err := getStrFromMetadata(fmt.Sprintf(gatewayURL, mac))
+	gw, err := utils.GetStrFromMetadata(fmt.Sprintf(gatewayURL, mac))
 	if err != nil {
 		return nil, err
 	}
@@ -276,27 +274,3 @@ func getNetConfFromMetadata(mac string) (*netConf, error) {
 	conf.routes = routes
 	return conf, nil
 }
-
-func getStrFromMetadata(url string) (string, error) {
-	resp, err := http.Get(url)
-	if err != nil {
-		return "", fmt.Errorf("error get url: %s from metaserver. %w", url, err)
-	}
-	//nolint:errcheck
-	defer resp.Body.Close()
-
-	if resp.StatusCode == http.StatusNotFound {
-		return "", fmt.Errorf("error get url: %s from metaserver, code: %v, %v", url, resp.StatusCode, "Not Found")
-	}
-	if resp.StatusCode >= http.StatusBadRequest {
-		return "", fmt.Errorf("error get url: %s from metaserver, code: %v", url, resp.StatusCode)
-	}
-
-	body, err := io.ReadAll(resp.Body)
-	if err != nil {
-		return "", err
-	}
-	result := strings.Split(string(body), "\n")
-	trimResult := strings.Trim(result[0], "/")
-	return trimResult, nil
-}

internal/utils/metadata.go

@@ -0,0 +1,186 @@
+package utils
+
+import (
+	"errors"
+	"fmt"
+	"io"
+	"net/http"
+	"strconv"
+	"strings"
+	"time"
+
+	"golang.org/x/sync/singleflight"
+	"k8s.io/apimachinery/pkg/util/cache"
+	"k8s.io/apimachinery/pkg/util/wait"
+)
+
+const (
+	tokenURL     = "http://100.100.100.200/latest/api/token"
+	tokenTimeout = 21600
+)
+
+func GetStrFromMetadata(url string) (string, error) {
+	body, err := getWithToken(url)
+	if err != nil {
+		return "", err
+	}
+	result := strings.Split(string(body), "\n")
+	trimResult := strings.Trim(result[0], "/")
+	return trimResult, nil
+}
+
+type Error struct {
+	URL  string
+	Code string
+	R    error
+}
+
+func (e *Error) Error() string {
+	return fmt.Sprintf("get from metaserver failed code: %s, url: %s, err: %s", e.Code, e.URL, e.R)
+}
+
+var (
+	tokenCache    *cache.Expiring
+	single        singleflight.Group
+	defaultClient *http.Client
+)
+
+func getWithToken(url string) ([]byte, error) {
+
+	skipRetry := false
+retry:
+	var token string
+	v, ok := tokenCache.Get(tokenURL)
+	if !ok {
+		vv, err, _ := single.Do(tokenURL, func() (interface{}, error) {
+			out, err := withRetry(tokenURL, [][]string{
+				{
+					"X-aliyun-ecs-metadata-token-ttl-seconds", strconv.Itoa(tokenTimeout),
+				},
+			})
+			if err != nil {
+				return nil, err
+			}
+			return string(out), nil
+		})
+		if err != nil {
+			return nil, err
+		}
+
+		token = vv.(string)
+
+		tokenCache.Set(tokenURL, token, tokenTimeout*time.Second/2)
+	} else {
+		token = v.(string)
+	}
+
+	out, err := withRetry(url, [][]string{
+		{
+			"X-aliyun-ecs-metadata-token", token,
+		},
+	})
+	if err != nil {
+		var typedErr *Error
+		ok := errors.As(err, &typedErr)
+
+		if ok && !skipRetry {
+			if typedErr.Code == strconv.Itoa(http.StatusUnauthorized) {
+				skipRetry = true
+
+				tokenCache.Delete(tokenURL)
+				goto retry
+			}
+		}
+
+		return nil, err
+	}
+
+	return out, err
+}
+
+func withRetry(url string, headers [][]string) ([]byte, error) {
+	var innerErr error
+	var body []byte
+	err := wait.ExponentialBackoff(wait.Backoff{
+		Duration: 500 * time.Millisecond,
+		Factor:   1.2,
+		Jitter:   0.1,
+		Steps:    4,
+	}, func() (bool, error) {
+		var err error
+
+		method := "GET"
+		if url == tokenURL {
+			method = "PUT"
+		}
+		req, err := http.NewRequest(method, url, nil)
+		if err != nil {
+			innerErr = &Error{
+				URL: url,
+				R:   err,
+			}
+			return false, nil
+		}
+
+		for _, h := range headers {
+			if len(h) != 2 {
+				return false, fmt.Errorf("invalid header")
+			}
+			req.Header.Set(h[0], h[1])
+		}
+
+		resp, err := defaultClient.Do(req)
+		if err != nil {
+			// retryable err
+			innerErr = &Error{
+				URL: url,
+				R:   err,
+			}
+			return false, nil
+		}
+		defer resp.Body.Close() // nolint:errcheck
+
+		// retryable err
+		if resp.StatusCode == http.StatusTooManyRequests ||
+			resp.StatusCode >= http.StatusInternalServerError {
+			innerErr = &Error{
+				URL:  url,
+				Code: strconv.Itoa(resp.StatusCode),
+				R:    nil,
+			}
+			return false, nil
+		}
+
+		if resp.StatusCode >= http.StatusBadRequest {
+			innerErr = &Error{
+				URL:  url,
+				Code: strconv.Itoa(resp.StatusCode),
+				R:    nil,
+			}
+			return false, innerErr
+		}
+
+		body, err = io.ReadAll(resp.Body)
+		if err != nil {
+			return false, err
+		}
+		return true, nil
+	})
+	if err != nil {
+		if innerErr != nil {
+			return nil, innerErr
+		}
+		return nil, err
+	}
+	return body, nil
+}
+
+func init() {
+	tokenCache = cache.NewExpiring()
+	defaultClient = &http.Client{
+		Transport:     nil,
+		CheckRedirect: nil,
+		Jar:           nil,
+		Timeout:       30 * time.Second,
+	}
+}