Merge pull request #13 from 30x/XAPID-608
Don’t panic on bad oauth and change server responses
diff --git a/apigeeSync_suite_test.go b/apigeeSync_suite_test.go
index be0f820..30f8dcb 100644
--- a/apigeeSync_suite_test.go
+++ b/apigeeSync_suite_test.go
@@ -51,17 +51,22 @@
config.Set(configConsumerKey, "XXXXXXX")
config.Set(configConsumerSecret, "YYYYYYY")
+ // fake an unreliable api server - always fails the first time
+ fail := false
+
// mock upstream testServer
testRouter.HandleFunc("/accesstoken", func(w http.ResponseWriter, req *http.Request) {
- defer GinkgoRecover()
+ // make unreliable
+ fail = !fail
+ if fail {
+ w.WriteHeader(500)
+ return
+ }
Expect(req.Method).To(Equal("POST"))
Expect(req.Header.Get("Content-Type")).To(Equal("application/x-www-form-urlencoded; param=value"))
err := req.ParseForm()
- // TODO: Test framework cannot handle this assertions and
- // this handler just stops and sends back ""
- // we need to handle it differently
Expect(err).NotTo(HaveOccurred())
Expect(req.Form.Get("grant_type")).To(Equal("client_credentials"))
Expect(req.Header.Get("status")).To(Equal("ONLINE"))
@@ -85,7 +90,12 @@
}).Methods("POST")
testRouter.HandleFunc("/snapshots", func(w http.ResponseWriter, req *http.Request) {
- defer GinkgoRecover()
+ // make unreliable
+ fail = !fail
+ if fail {
+ w.WriteHeader(500)
+ return
+ }
q := req.URL.Query()
@@ -198,7 +208,11 @@
}).Methods("GET")
testRouter.HandleFunc("/changes", func(w http.ResponseWriter, req *http.Request) {
- defer GinkgoRecover()
+ fail = !fail
+ if fail {
+ w.WriteHeader(500)
+ return
+ }
if req.URL.Query().Get("since") == "lastSeq_01" {
go func() {
@@ -270,8 +284,6 @@
// This is actually the first test :)
// Tests that entire bootstrap and set of sync operations work
apid.Events().ListenFunc(ApigeeSyncEventSelector, func(event apid.Event) {
- defer GinkgoRecover()
-
if s, ok := event.(*common.Snapshot); ok {
Expect(s.SnapshotInfo).Should(Equal("snapinfo1"))
diff --git a/apigee_sync.go b/apigee_sync.go
index 44845e6..1e0a04a 100644
--- a/apigee_sync.go
+++ b/apigee_sync.go
@@ -109,11 +109,8 @@
for {
log.Debug("polling...")
if token == "" {
- /* token not valid?, get a new token */
- status := getBearerToken()
- if status == false {
- return errors.New("Unable to get new token")
- }
+ // invalid token, loop until we get one
+ getBearerToken()
}
/* Find the scopes associated with the config id */
@@ -213,67 +210,95 @@
}
}
+
+// simple doubling back-off
+func createBackOff(retryIn, maxBackOff time.Duration) func() {
+ return func() {
+ log.Debugf("backoff called. will retry in %s.", retryIn)
+ time.Sleep(retryIn)
+ retryIn = retryIn * time.Duration(2)
+ if retryIn > maxBackOff {
+ retryIn = maxBackOff
+ }
+ }
+}
+
/*
* This function will (for now) use the Access Key/Secret Key/ApidConfig Id
* to get the bearer token, and the scopes (as comma separated scope)
*/
-func getBearerToken() bool {
+func getBearerToken() {
log.Info("Getting a Bearer token...")
- uri, err := url.Parse(config.GetString(configProxyServerBaseURI))
+ uriString := config.GetString(configProxyServerBaseURI)
+ uri, err := url.Parse(uriString)
if err != nil {
- log.Error(err)
- return false
+ log.Panicf("unable to parse uri config '%s' value: '%s': %v", configProxyServerBaseURI, uriString, err)
}
uri.Path = path.Join(uri.Path, "/accesstoken")
- token = ""
- form := url.Values{}
- form.Set("grant_type", "client_credentials")
- form.Add("client_id", config.GetString(configConsumerKey))
- form.Add("client_secret", config.GetString(configConsumerSecret))
- req, err := http.NewRequest("POST", uri.String(), bytes.NewBufferString(form.Encode()))
- req.Header.Set("Content-Type", "application/x-www-form-urlencoded; param=value")
- req.Header.Set("display_name", apidInfo.InstanceName)
- req.Header.Set("apid_instance_id", apidInfo.InstanceID)
- req.Header.Set("apid_cluster_Id", apidInfo.ClusterID)
- req.Header.Set("status", "ONLINE")
- req.Header.Set("created_at_apid", time.Now().Format(time.RFC3339))
- req.Header.Set("plugin_details", apidPluginDetails)
+ retryIn := 5 * time.Millisecond
+ maxBackOff := 1 * time.Minute
+ backOffFunc := createBackOff(retryIn, maxBackOff)
+ first := true
- client := &http.Client{}
- resp, err := client.Do(req)
- if err != nil {
- log.Error("Unable to Connect to Edge Proxy Server ", err)
- return false
- }
- defer resp.Body.Close()
- if resp.StatusCode != 200 {
- log.Error("Oauth Request Failed with Resp Code ", resp.StatusCode)
- return false
- }
- body, err := ioutil.ReadAll(resp.Body)
- if err != nil {
- log.Error("Unable to read EdgeProxy Sever response ", err)
- return false
- }
+ for {
+ if first {
+ first = false
+ } else {
+ backOffFunc()
+ }
- var oauthResp oauthTokenResp
- log.Debugf("Response: %s ", body)
- err = json.Unmarshal(body, &oauthResp)
- if err != nil {
- log.Error(err)
- return false
- }
- token = oauthResp.AccessToken
+ token = ""
+ form := url.Values{}
+ form.Set("grant_type", "client_credentials")
+ form.Add("client_id", config.GetString(configConsumerKey))
+ form.Add("client_secret", config.GetString(configConsumerSecret))
+ req, err := http.NewRequest("POST", uri.String(), bytes.NewBufferString(form.Encode()))
+ req.Header.Set("Content-Type", "application/x-www-form-urlencoded; param=value")
+ req.Header.Set("display_name", apidInfo.InstanceName)
+ req.Header.Set("apid_instance_id", apidInfo.InstanceID)
+ req.Header.Set("apid_cluster_Id", apidInfo.ClusterID)
+ req.Header.Set("status", "ONLINE")
+ req.Header.Set("created_at_apid", time.Now().Format(time.RFC3339))
+ req.Header.Set("plugin_details", apidPluginDetails)
- /*
- * This stores the bearer token for any other plugin to
- * consume.
- */
- config.Set(bearerToken, token)
- log.Debug("Got a new Bearer token.")
- return true
+ client := &http.Client{}
+ resp, err := client.Do(req)
+ if err != nil {
+ log.Errorf("Unable to Connect to Edge Proxy Server: %v", err)
+ continue
+ }
+ defer resp.Body.Close()
+ if resp.StatusCode != 200 {
+ log.Errorf("Oauth Request Failed with Resp Code: %v", resp.StatusCode)
+ continue
+ }
+ body, err := ioutil.ReadAll(resp.Body)
+ if err != nil {
+ log.Errorf("Unable to read EdgeProxy Sever response: %v", err)
+ continue
+ }
+
+ var oauthResp oauthTokenResp
+ log.Debugf("Response: %s ", body)
+ err = json.Unmarshal(body, &oauthResp)
+ if err != nil {
+ log.Error("unable to unmarshal JSON response %s: %v", string(body), err)
+ continue
+ }
+ token = oauthResp.AccessToken
+
+ /*
+ * This stores the bearer token for any other plugin to
+ * consume.
+ */
+ config.Set(bearerToken, token)
+
+ log.Debug("Got a new Bearer token.")
+
+ return
+ }
}
type oauthTokenResp struct {
@@ -365,16 +390,15 @@
log.Debugf("downloadSnapshot")
- /* Get the bearer token */
- status := getBearerToken()
- if status == false {
- log.Errorf("Unable to get Bearer token or is Invalid")
- }
snapshotUri, err := url.Parse(config.GetString(configSnapServerBaseURI))
if err != nil {
- log.Fatalf("bad url value for config %s: %s", snapshotUri, err)
+ log.Panicf("bad url value for config %s: %s", snapshotUri, err)
}
+ // getBearerToken loops until good
+ getBearerToken()
+ // todo: this could expire... ensure it's called again as needed
+
var scopes []string
if downloadBootSnapshot {
scopes = findScopesForId(apidInfo.ClusterID)
@@ -397,47 +421,57 @@
client := &http.Client{
CheckRedirect: Redirect,
}
- req, err := http.NewRequest("GET", uri, nil)
- addHeaders(req)
- /* Set the transport protocol type based on conf file input */
- if config.GetString(configSnapshotProtocol) == "json" {
- req.Header.Set("Accept", "application/json")
- } else {
- req.Header.Set("Accept", "application/proto")
- }
-
- /* Issue the request to the snapshot server */
- r, err := client.Do(req)
- if err != nil {
- log.Fatalf("Snapshotserver comm error: %v", err)
- }
- defer r.Body.Close()
-
- /* Decode the Snapshot server response */
- var resp common.Snapshot
- err = json.NewDecoder(r.Body).Decode(&resp)
- if err != nil {
-
- if downloadBootSnapshot {
- /*
- * If the data set is empty, allow it to proceed, as change server
- * will feed data. Since Bootstrapping has passed, it has the
- * Bootstrap config id to function.
- */
- downloadDataSnapshot = true
- return
- } else {
- log.Fatalf("JSON Response Data not parsable: %v", err)
+ for {
+ req, err := http.NewRequest("GET", uri, nil)
+ if err != nil {
+ // should never happen, but if it does, it's unrecoverable anyway
+ log.Panicf("Snapshotserver comm error: %v", err)
}
- }
+ addHeaders(req)
- if r.StatusCode == 200 {
+ // Set the transport protocol type based on conf file input
+ if config.GetString(configSnapshotProtocol) == "json" {
+ req.Header.Set("Accept", "application/json")
+ } else {
+ req.Header.Set("Accept", "application/proto")
+ }
+
+ // Issue the request to the snapshot server
+ r, err := client.Do(req)
+ if err != nil {
+ log.Errorf("Snapshotserver comm error: %v", err)
+ continue
+ }
+
+ // Decode the Snapshot server response
+ var resp common.Snapshot
+ err = json.NewDecoder(r.Body).Decode(&resp)
+ r.Body.Close()
+ if err != nil {
+ if downloadBootSnapshot {
+ /*
+ * If the data set is empty, allow it to proceed, as change server
+ * will feed data. Since Bootstrapping has passed, it has the
+ * Bootstrap config id to function.
+ */
+ downloadDataSnapshot = true
+ return
+ } else {
+ log.Errorf("JSON Response Data not parsable: %v", err)
+ continue
+ }
+ }
+
+ if r.StatusCode != 200 {
+ log.Errorf("Snapshot server conn failed. HTTP Resp code %d", r.StatusCode)
+ continue
+ }
+
log.Info("Emitting Snapshot to plugins")
events.ListenFunc(apid.EventDeliveredSelector, postPluginDataDelivery)
events.Emit(ApigeeSyncEventSelector, &resp)
- } else {
- log.Fatalf("Snapshot server conn failed. HTTP Resp code %d", r.StatusCode)
+ break
}
}
