Merge pull request #123 from c4milo/shared-http-client
feat(util/http_client): Adds generic HTTP client
This commit is contained in:
commit
1a295f65c7
2
.gitignore
vendored
2
.gitignore
vendored
@ -1,4 +1,4 @@
|
|||||||
*.swp
|
*.swp
|
||||||
bin/
|
bin/
|
||||||
coverage/
|
coverage/
|
||||||
pkg/
|
gopath/
|
||||||
|
7
build
7
build
@ -3,7 +3,12 @@
|
|||||||
ORG_PATH="github.com/coreos"
|
ORG_PATH="github.com/coreos"
|
||||||
REPO_PATH="${ORG_PATH}/coreos-cloudinit"
|
REPO_PATH="${ORG_PATH}/coreos-cloudinit"
|
||||||
|
|
||||||
|
if [ ! -h gopath/src/${REPO_PATH} ]; then
|
||||||
|
mkdir -p gopath/src/${ORG_PATH}
|
||||||
|
ln -s ../../../.. gopath/src/${REPO_PATH} || exit 255
|
||||||
|
fi
|
||||||
|
|
||||||
export GOBIN=${PWD}/bin
|
export GOBIN=${PWD}/bin
|
||||||
export GOPATH=${PWD}
|
export GOPATH=${PWD}/gopath
|
||||||
|
|
||||||
go build -o bin/coreos-cloudinit ${REPO_PATH}
|
go build -o bin/coreos-cloudinit ${REPO_PATH}
|
||||||
|
@ -1,104 +1,6 @@
|
|||||||
package datasource
|
package datasource
|
||||||
|
|
||||||
import (
|
|
||||||
"errors"
|
|
||||||
"fmt"
|
|
||||||
"io/ioutil"
|
|
||||||
"log"
|
|
||||||
"math"
|
|
||||||
"net"
|
|
||||||
"net/http"
|
|
||||||
neturl "net/url"
|
|
||||||
"strings"
|
|
||||||
"time"
|
|
||||||
)
|
|
||||||
|
|
||||||
const (
|
|
||||||
HTTP_2xx = 2
|
|
||||||
HTTP_4xx = 4
|
|
||||||
|
|
||||||
maxTimeout = time.Second * 5
|
|
||||||
maxRetries = 15
|
|
||||||
)
|
|
||||||
|
|
||||||
type Datasource interface {
|
type Datasource interface {
|
||||||
Fetch() ([]byte, error)
|
Fetch() ([]byte, error)
|
||||||
Type() string
|
Type() string
|
||||||
}
|
}
|
||||||
|
|
||||||
// HTTP client timeout
|
|
||||||
// This one is low since exponential backoff will kick off too.
|
|
||||||
var timeout = time.Duration(2) * time.Second
|
|
||||||
|
|
||||||
func dialTimeout(network, addr string) (net.Conn, error) {
|
|
||||||
deadline := time.Now().Add(timeout)
|
|
||||||
c, err := net.DialTimeout(network, addr, timeout)
|
|
||||||
if err != nil {
|
|
||||||
return nil, err
|
|
||||||
}
|
|
||||||
c.SetDeadline(deadline)
|
|
||||||
return c, nil
|
|
||||||
}
|
|
||||||
|
|
||||||
// Fetches user-data url with support for exponential backoff and maximum retries
|
|
||||||
func fetchURL(rawurl string) ([]byte, error) {
|
|
||||||
if rawurl == "" {
|
|
||||||
return nil, errors.New("user-data URL is empty. Skipping.")
|
|
||||||
}
|
|
||||||
|
|
||||||
url, err := neturl.Parse(rawurl)
|
|
||||||
if err != nil {
|
|
||||||
return nil, err
|
|
||||||
}
|
|
||||||
|
|
||||||
// Unfortunately, url.Parse is too generic to throw errors if a URL does not
|
|
||||||
// have a valid HTTP scheme. So, we have to do this extra validation
|
|
||||||
if !strings.HasPrefix(url.Scheme, "http") {
|
|
||||||
return nil, fmt.Errorf("user-data URL %s does not have a valid HTTP scheme. Skipping.", rawurl)
|
|
||||||
}
|
|
||||||
|
|
||||||
userdataURL := url.String()
|
|
||||||
|
|
||||||
// We need to create our own client in order to add timeout support.
|
|
||||||
// TODO(c4milo) Replace it once Go 1.3 is officially used by CoreOS
|
|
||||||
// More info: https://code.google.com/p/go/source/detail?r=ada6f2d5f99f
|
|
||||||
transport := &http.Transport{
|
|
||||||
Dial: dialTimeout,
|
|
||||||
}
|
|
||||||
|
|
||||||
client := &http.Client{
|
|
||||||
Transport: transport,
|
|
||||||
}
|
|
||||||
|
|
||||||
for retry := 1; retry <= maxRetries; retry++ {
|
|
||||||
log.Printf("Fetching user-data from %s. Attempt #%d", userdataURL, retry)
|
|
||||||
|
|
||||||
resp, err := client.Get(userdataURL)
|
|
||||||
|
|
||||||
if err == nil {
|
|
||||||
defer resp.Body.Close()
|
|
||||||
status := resp.StatusCode / 100
|
|
||||||
|
|
||||||
if status == HTTP_2xx {
|
|
||||||
return ioutil.ReadAll(resp.Body)
|
|
||||||
}
|
|
||||||
|
|
||||||
if status == HTTP_4xx {
|
|
||||||
return nil, fmt.Errorf("user-data not found. HTTP status code: %d", resp.StatusCode)
|
|
||||||
}
|
|
||||||
|
|
||||||
log.Printf("user-data not found. HTTP status code: %d", resp.StatusCode)
|
|
||||||
} else {
|
|
||||||
log.Printf("unable to fetch user-data: %s", err.Error())
|
|
||||||
}
|
|
||||||
|
|
||||||
duration := time.Millisecond * time.Duration((math.Pow(float64(2), float64(retry)) * 100))
|
|
||||||
if duration > maxTimeout {
|
|
||||||
duration = maxTimeout
|
|
||||||
}
|
|
||||||
|
|
||||||
time.Sleep(duration)
|
|
||||||
}
|
|
||||||
|
|
||||||
return nil, fmt.Errorf("unable to fetch user-data. Maximum retries reached: %d", maxRetries)
|
|
||||||
}
|
|
||||||
|
@ -1,5 +1,7 @@
|
|||||||
package datasource
|
package datasource
|
||||||
|
|
||||||
|
import "github.com/coreos/coreos-cloudinit/pkg"
|
||||||
|
|
||||||
type metadataService struct {
|
type metadataService struct {
|
||||||
url string
|
url string
|
||||||
}
|
}
|
||||||
@ -9,7 +11,8 @@ func NewMetadataService(url string) *metadataService {
|
|||||||
}
|
}
|
||||||
|
|
||||||
func (ms *metadataService) Fetch() ([]byte, error) {
|
func (ms *metadataService) Fetch() ([]byte, error) {
|
||||||
return fetchURL(ms.url)
|
client := pkg.NewHttpClient()
|
||||||
|
return client.Get(ms.url)
|
||||||
}
|
}
|
||||||
|
|
||||||
func (ms *metadataService) Type() string {
|
func (ms *metadataService) Type() string {
|
||||||
|
@ -5,6 +5,8 @@ import (
|
|||||||
"io/ioutil"
|
"io/ioutil"
|
||||||
"log"
|
"log"
|
||||||
"strings"
|
"strings"
|
||||||
|
|
||||||
|
"github.com/coreos/coreos-cloudinit/pkg"
|
||||||
)
|
)
|
||||||
|
|
||||||
const (
|
const (
|
||||||
@ -32,7 +34,8 @@ func (self *procCmdline) Fetch() ([]byte, error) {
|
|||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
|
|
||||||
cfg, err := fetchURL(url)
|
client := pkg.NewHttpClient()
|
||||||
|
cfg, err := client.Get(url)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
|
@ -3,9 +3,8 @@ package initialize
|
|||||||
import (
|
import (
|
||||||
"encoding/json"
|
"encoding/json"
|
||||||
"fmt"
|
"fmt"
|
||||||
"io/ioutil"
|
|
||||||
"net/http"
|
|
||||||
|
|
||||||
|
"github.com/coreos/coreos-cloudinit/pkg"
|
||||||
"github.com/coreos/coreos-cloudinit/system"
|
"github.com/coreos/coreos-cloudinit/system"
|
||||||
)
|
)
|
||||||
|
|
||||||
@ -25,22 +24,19 @@ func SSHImportKeysFromURL(system_user string, url string) error {
|
|||||||
}
|
}
|
||||||
|
|
||||||
func fetchUserKeys(url string) ([]string, error) {
|
func fetchUserKeys(url string) ([]string, error) {
|
||||||
res, err := http.Get(url)
|
client := pkg.NewHttpClient()
|
||||||
|
data, err := client.Get(url)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
defer res.Body.Close()
|
|
||||||
body, err := ioutil.ReadAll(res.Body)
|
var userKeys []UserKey
|
||||||
if err != nil {
|
err = json.Unmarshal(data, &userKeys)
|
||||||
return nil, err
|
|
||||||
}
|
|
||||||
var data []UserKey
|
|
||||||
err = json.Unmarshal(body, &data)
|
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return nil, err
|
return nil, err
|
||||||
}
|
}
|
||||||
keys := make([]string, 0)
|
keys := make([]string, 0)
|
||||||
for _, key := range data {
|
for _, key := range userKeys {
|
||||||
keys = append(keys, key.Key)
|
keys = append(keys, key.Key)
|
||||||
}
|
}
|
||||||
return keys, err
|
return keys, err
|
||||||
|
120
pkg/http_client.go
Normal file
120
pkg/http_client.go
Normal file
@ -0,0 +1,120 @@
|
|||||||
|
package pkg
|
||||||
|
|
||||||
|
import (
|
||||||
|
"crypto/tls"
|
||||||
|
"errors"
|
||||||
|
"fmt"
|
||||||
|
"io/ioutil"
|
||||||
|
"log"
|
||||||
|
"math"
|
||||||
|
"net"
|
||||||
|
"net/http"
|
||||||
|
neturl "net/url"
|
||||||
|
"strings"
|
||||||
|
"time"
|
||||||
|
)
|
||||||
|
|
||||||
|
const (
|
||||||
|
HTTP_2xx = 2
|
||||||
|
HTTP_4xx = 4
|
||||||
|
)
|
||||||
|
|
||||||
|
type HttpClient struct {
|
||||||
|
// Maximum exp backoff duration. Defaults to 5 seconds
|
||||||
|
MaxBackoff time.Duration
|
||||||
|
|
||||||
|
// Maximum number of connection retries. Defaults to 15
|
||||||
|
MaxRetries int
|
||||||
|
|
||||||
|
// HTTP client timeout, this is suggested to be low since exponential
|
||||||
|
// backoff will kick off too. Defaults to 2 seconds
|
||||||
|
Timeout time.Duration
|
||||||
|
|
||||||
|
// Whether or not to skip TLS verification. Defaults to false
|
||||||
|
SkipTLS bool
|
||||||
|
}
|
||||||
|
|
||||||
|
func NewHttpClient() *HttpClient {
|
||||||
|
return &HttpClient{
|
||||||
|
MaxBackoff: time.Second * 5,
|
||||||
|
MaxRetries: 15,
|
||||||
|
Timeout: time.Duration(2) * time.Second,
|
||||||
|
SkipTLS: false,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
// Fetches a given URL with support for exponential backoff and maximum retries
|
||||||
|
func (h *HttpClient) Get(rawurl string) ([]byte, error) {
|
||||||
|
if rawurl == "" {
|
||||||
|
return nil, errors.New("URL is empty. Skipping.")
|
||||||
|
}
|
||||||
|
|
||||||
|
url, err := neturl.Parse(rawurl)
|
||||||
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
|
||||||
|
// Unfortunately, url.Parse is too generic to throw errors if a URL does not
|
||||||
|
// have a valid HTTP scheme. So, we have to do this extra validation
|
||||||
|
if !strings.HasPrefix(url.Scheme, "http") {
|
||||||
|
return nil, fmt.Errorf("URL %s does not have a valid HTTP scheme. Skipping.", rawurl)
|
||||||
|
}
|
||||||
|
|
||||||
|
dataURL := url.String()
|
||||||
|
|
||||||
|
// We need to create our own client in order to add timeout support.
|
||||||
|
// TODO(c4milo) Replace it once Go 1.3 is officially used by CoreOS
|
||||||
|
// More info: https://code.google.com/p/go/source/detail?r=ada6f2d5f99f
|
||||||
|
transport := &http.Transport{
|
||||||
|
TLSClientConfig: &tls.Config{
|
||||||
|
InsecureSkipVerify: h.SkipTLS,
|
||||||
|
},
|
||||||
|
Dial: func(network, addr string) (net.Conn, error) {
|
||||||
|
deadline := time.Now().Add(h.Timeout)
|
||||||
|
c, err := net.DialTimeout(network, addr, h.Timeout)
|
||||||
|
if err != nil {
|
||||||
|
return nil, err
|
||||||
|
}
|
||||||
|
c.SetDeadline(deadline)
|
||||||
|
return c, nil
|
||||||
|
},
|
||||||
|
}
|
||||||
|
|
||||||
|
client := &http.Client{
|
||||||
|
Transport: transport,
|
||||||
|
}
|
||||||
|
|
||||||
|
for retry := 1; retry <= h.MaxRetries; retry++ {
|
||||||
|
log.Printf("Fetching data from %s. Attempt #%d", dataURL, retry)
|
||||||
|
|
||||||
|
resp, err := client.Get(dataURL)
|
||||||
|
|
||||||
|
if err == nil {
|
||||||
|
defer resp.Body.Close()
|
||||||
|
status := resp.StatusCode / 100
|
||||||
|
|
||||||
|
if status == HTTP_2xx {
|
||||||
|
return ioutil.ReadAll(resp.Body)
|
||||||
|
}
|
||||||
|
|
||||||
|
if status == HTTP_4xx {
|
||||||
|
return nil, fmt.Errorf("Not found. HTTP status code: %d", resp.StatusCode)
|
||||||
|
}
|
||||||
|
|
||||||
|
log.Printf("Server error. HTTP status code: %d", resp.StatusCode)
|
||||||
|
} else {
|
||||||
|
log.Printf("Unable to fetch data: %s", err.Error())
|
||||||
|
}
|
||||||
|
|
||||||
|
duration := time.Millisecond * time.Duration((math.Pow(float64(2), float64(retry)) * 100))
|
||||||
|
if duration > h.MaxBackoff {
|
||||||
|
duration = h.MaxBackoff
|
||||||
|
}
|
||||||
|
|
||||||
|
log.Printf("Sleeping for %v...", duration)
|
||||||
|
|
||||||
|
time.Sleep(duration)
|
||||||
|
}
|
||||||
|
|
||||||
|
return nil, fmt.Errorf("Unable to fetch data. Maximum retries reached: %d", h.MaxRetries)
|
||||||
|
}
|
@ -1,4 +1,4 @@
|
|||||||
package datasource
|
package pkg
|
||||||
|
|
||||||
import (
|
import (
|
||||||
"fmt"
|
"fmt"
|
||||||
@ -19,7 +19,9 @@ var expBackoffTests = []struct {
|
|||||||
|
|
||||||
// Test exponential backoff and that it continues retrying if a 5xx response is
|
// Test exponential backoff and that it continues retrying if a 5xx response is
|
||||||
// received
|
// received
|
||||||
func TestFetchURLExpBackOff(t *testing.T) {
|
func TestGetURLExpBackOff(t *testing.T) {
|
||||||
|
client := NewHttpClient()
|
||||||
|
|
||||||
for i, tt := range expBackoffTests {
|
for i, tt := range expBackoffTests {
|
||||||
mux := http.NewServeMux()
|
mux := http.NewServeMux()
|
||||||
count := 0
|
count := 0
|
||||||
@ -34,7 +36,7 @@ func TestFetchURLExpBackOff(t *testing.T) {
|
|||||||
ts := httptest.NewServer(mux)
|
ts := httptest.NewServer(mux)
|
||||||
defer ts.Close()
|
defer ts.Close()
|
||||||
|
|
||||||
data, err := fetchURL(ts.URL)
|
data, err := client.Get(ts.URL)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
t.Errorf("Test case %d produced error: %v", i, err)
|
t.Errorf("Test case %d produced error: %v", i, err)
|
||||||
}
|
}
|
||||||
@ -50,7 +52,8 @@ func TestFetchURLExpBackOff(t *testing.T) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
// Test that it stops retrying if a 4xx response comes back
|
// Test that it stops retrying if a 4xx response comes back
|
||||||
func TestFetchURL4xx(t *testing.T) {
|
func TestGetURL4xx(t *testing.T) {
|
||||||
|
client := NewHttpClient()
|
||||||
retries := 0
|
retries := 0
|
||||||
ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||||
retries++
|
retries++
|
||||||
@ -58,9 +61,9 @@ func TestFetchURL4xx(t *testing.T) {
|
|||||||
}))
|
}))
|
||||||
defer ts.Close()
|
defer ts.Close()
|
||||||
|
|
||||||
_, err := fetchURL(ts.URL)
|
_, err := client.Get(ts.URL)
|
||||||
if err == nil {
|
if err == nil {
|
||||||
t.Errorf("Incorrect result\ngot: %s\nwant: %s", err.Error(), "user-data not found. HTTP status code: 404")
|
t.Errorf("Incorrect result\ngot: %s\nwant: %s", err.Error(), "Not found. HTTP status code: 404")
|
||||||
}
|
}
|
||||||
|
|
||||||
if retries > 1 {
|
if retries > 1 {
|
||||||
@ -69,7 +72,7 @@ func TestFetchURL4xx(t *testing.T) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
// Test that it fetches and returns user-data just fine
|
// Test that it fetches and returns user-data just fine
|
||||||
func TestFetchURL2xx(t *testing.T) {
|
func TestGetURL2xx(t *testing.T) {
|
||||||
var cloudcfg = `
|
var cloudcfg = `
|
||||||
#cloud-config
|
#cloud-config
|
||||||
coreos:
|
coreos:
|
||||||
@ -83,12 +86,13 @@ coreos:
|
|||||||
reboot-strategy: best-effort
|
reboot-strategy: best-effort
|
||||||
`
|
`
|
||||||
|
|
||||||
|
client := NewHttpClient()
|
||||||
ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||||
fmt.Fprint(w, cloudcfg)
|
fmt.Fprint(w, cloudcfg)
|
||||||
}))
|
}))
|
||||||
defer ts.Close()
|
defer ts.Close()
|
||||||
|
|
||||||
data, err := fetchURL(ts.URL)
|
data, err := client.Get(ts.URL)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
t.Errorf("Incorrect result\ngot: %v\nwant: %v", err, nil)
|
t.Errorf("Incorrect result\ngot: %v\nwant: %v", err, nil)
|
||||||
}
|
}
|
||||||
@ -99,19 +103,21 @@ coreos:
|
|||||||
}
|
}
|
||||||
|
|
||||||
// Test attempt to fetching using malformed URL
|
// Test attempt to fetching using malformed URL
|
||||||
func TestFetchURLMalformed(t *testing.T) {
|
func TestGetMalformedURL(t *testing.T) {
|
||||||
|
client := NewHttpClient()
|
||||||
|
|
||||||
var tests = []struct {
|
var tests = []struct {
|
||||||
url string
|
url string
|
||||||
want string
|
want string
|
||||||
}{
|
}{
|
||||||
{"boo", "user-data URL boo does not have a valid HTTP scheme. Skipping."},
|
{"boo", "URL boo does not have a valid HTTP scheme. Skipping."},
|
||||||
{"mailto://boo", "user-data URL mailto://boo does not have a valid HTTP scheme. Skipping."},
|
{"mailto://boo", "URL mailto://boo does not have a valid HTTP scheme. Skipping."},
|
||||||
{"ftp://boo", "user-data URL ftp://boo does not have a valid HTTP scheme. Skipping."},
|
{"ftp://boo", "URL ftp://boo does not have a valid HTTP scheme. Skipping."},
|
||||||
{"", "user-data URL is empty. Skipping."},
|
{"", "URL is empty. Skipping."},
|
||||||
}
|
}
|
||||||
|
|
||||||
for _, test := range tests {
|
for _, test := range tests {
|
||||||
_, err := fetchURL(test.url)
|
_, err := client.Get(test.url)
|
||||||
if err == nil || err.Error() != test.want {
|
if err == nil || err.Error() != test.want {
|
||||||
t.Errorf("Incorrect result\ngot: %v\nwant: %v", err, test.want)
|
t.Errorf("Incorrect result\ngot: %v\nwant: %v", err, test.want)
|
||||||
}
|
}
|
@ -1 +0,0 @@
|
|||||||
../../../
|
|
2
test
2
test
@ -13,7 +13,7 @@ COVER=${COVER:-"-cover"}
|
|||||||
|
|
||||||
source ./build
|
source ./build
|
||||||
|
|
||||||
declare -a TESTPKGS=(initialize system datasource)
|
declare -a TESTPKGS=(initialize system datasource pkg)
|
||||||
|
|
||||||
if [ -z "$PKG" ]; then
|
if [ -z "$PKG" ]; then
|
||||||
GOFMTPATH="$TESTPKGS coreos-cloudinit.go"
|
GOFMTPATH="$TESTPKGS coreos-cloudinit.go"
|
||||||
|
Loading…
x
Reference in New Issue
Block a user