pipeline/mutator: Refactor hydrator retry config (#287)

.schemas/config.schema.json

@@ -637,15 +637,15 @@
               "type": "object",
               "additionalProperties": false,
               "properties": {
-                "number_of_retries": {
-                  "type": "number",
-                  "minimum": 0,
-                  "default": 100
+                "give_up_after": {
+                  "type": "string",
+                  "default": "1s",
+                  "pattern": "^[0-9]+(ns|us|ms|s|m|h)$"
                 },
-                "delay_in_milliseconds": {
-                  "type": "number",
-                  "minimum": 0,
-                  "default": 3
+                "max_delay": {
+                  "type": "string",
+                  "pattern": "^[0-9]+(ns|us|ms|s|m|h)$",
+                  "default": "100ms"
                 }
               }
             }

UPGRADE.md

@@ -44,6 +44,40 @@ before finalizing the upgrade process.
 
 ## master
 
+## v0.33.0-beta.1+oryOS.13
+
+The "mutator" hydrator config has changed:
+
+- `config.retry.delay_in_milliseconds: 100` (int) is now
+  `config.retry.max_delay: 100ms` (duration)
+- `config.retry.max_retries: 3` (int) is now `config.retry.give_up_after: 1s`
+  (duration)
+
+A new feature introduce in this release allows to keep using existing access
+rules by setting `"version": "v0.32.0-beta.1"` in the existing rules. ORY
+Oathkeeper will migrate the old config to the new config.
+
+This access rule definition will properly be migrated:
+
+```
+{
++  "version": "v0.32.0-beta.1",
+  "mutators": [
+    {
+      "handler": "hydrator",
+      "config": {
+        "retry": {
+          "delay_in_milliseconds": 500,
+          "max_retries": 5
+        }
+      }
+    }
+  ]
+}
+```
+
+We encourage you to tag all your access rules with the version.
+
 ## v0.32.0-beta.1+oryOS.12
 
 An issue with the release pipeline has been resolved, which required several

driver/configuration/provider_viper_public_test.go

@@ -21,6 +21,7 @@ import (
 	"github.com/ory/oathkeeper/pipeline/authn"
 	"github.com/ory/oathkeeper/pipeline/authz"
 	"github.com/ory/oathkeeper/pipeline/mutate"
+	"github.com/ory/oathkeeper/x"
 )
 
 func TestPipelineConfig(t *testing.T) {
@@ -56,9 +57,9 @@ func TestPipelineConfig(t *testing.T) {
 		require.NoError(t, p.PipelineConfig("mutators", "hydrator", json.RawMessage(``), &dec))
 		assert.Equal(t, "https://some-url/", dec.Api.URL)
 
-		require.NoError(t, p.PipelineConfig("mutators", "hydrator", json.RawMessage(`{"api":{"url":"http://override-url/foo","retry":{"number_of_retries":15}}}`), &dec))
+		require.NoError(t, p.PipelineConfig("mutators", "hydrator", json.RawMessage(`{"api":{"url":"http://override-url/foo","retry":{"give_up_after":"15s"}}}`), &dec))
 		assert.Equal(t, "http://override-url/foo", dec.Api.URL)
-		assert.Equal(t, 15, dec.Api.Retry.NumberOfRetries)
+		assert.Equal(t, "15s", dec.Api.Retry.GiveUpAfter)
 	})
 
 	t.Run("case=should pass array values", func(t *testing.T) {
@@ -267,7 +268,7 @@ func TestViperProvider(t *testing.T) {
 		})
 
 		t.Run("mutator=hydrator", func(t *testing.T) {
-			a := mutate.NewMutatorHydrator(p)
+			a := mutate.NewMutatorHydrator(p, new(x.TestLoggerProvider))
 			assert.True(t, p.MutatorIsEnabled(a.GetID()))
 			require.NoError(t, a.Validate(nil))
 		})

driver/registry_memory.go

@@ -335,7 +335,7 @@ func (r *RegistryMemory) prepareMutators() {
 			mutate.NewMutatorHeader(r.c),
 			mutate.NewMutatorIDToken(r.c, r),
 			mutate.NewMutatorNoop(r.c),
-			mutate.NewMutatorHydrator(r.c),
+			mutate.NewMutatorHydrator(r.c, r),
 		}
 
 		r.mutators = map[string]mutate.Mutator{}

pipeline/mutate/mutator_hydrator.go

@@ -28,8 +28,7 @@ import (
 	"time"
 
 	"github.com/ory/oathkeeper/pipeline/authn"
-
-	"github.com/cenkalti/backoff"
+	"github.com/ory/oathkeeper/x"
 
 	"github.com/ory/x/httpx"
 
@@ -55,6 +54,7 @@ const (
 type MutatorHydrator struct {
 	c      configuration.Provider
 	client *http.Client
+	d      mutatorHydratorDependencies
 }
 
 type BasicAuth struct {
@@ -67,8 +67,8 @@ type auth struct {
 }
 
 type retryConfig struct {
-	NumberOfRetries     int `json:"number_of_retries"`
-	DelayInMilliseconds int `json:"delay_in_milliseconds"`
+	MaxDelay    string `json:"max_delay"`
+	GiveUpAfter string `json:"give_up_after"`
 }
 
 type externalAPIConfig struct {
@@ -81,8 +81,12 @@ type MutatorHydratorConfig struct {
 	Api externalAPIConfig `json:"api"`
 }
 
-func NewMutatorHydrator(c configuration.Provider) *MutatorHydrator {
-	return &MutatorHydrator{c: c, client: httpx.NewResilientClientLatencyToleranceSmall(nil)}
+type mutatorHydratorDependencies interface {
+	x.RegistryLogger
+}
+
+func NewMutatorHydrator(c configuration.Provider, d mutatorHydratorDependencies) *MutatorHydrator {
+	return &MutatorHydrator{c: c, d: d, client: httpx.NewResilientClientLatencyToleranceSmall(nil)}
 }
 
 func (a *MutatorHydrator) GetID() string {
@@ -120,31 +124,44 @@ func (a *MutatorHydrator) Mutate(r *http.Request, session *authn.AuthenticationS
 	}
 	req.Header.Set(contentTypeHeaderKey, contentTypeJSONHeaderValue)
 
-	retryConfig := retryConfig{defaultNumberOfRetries, defaultDelayInMilliseconds}
+	var client http.Client
 	if cfg.Api.Retry != nil {
-		retryConfig = *cfg.Api.Retry
-	}
-	var res *http.Response
-	err = backoff.Retry(func() error {
-		res, err = a.client.Do(req)
-		if err != nil {
-			return errors.WithStack(err)
+		maxRetryDelay := time.Second
+		giveUpAfter := time.Millisecond * 50
+		if len(cfg.Api.Retry.MaxDelay) > 0 {
+			if d, err := time.ParseDuration(cfg.Api.Retry.MaxDelay); err != nil {
+				a.d.Logger().WithError(err).Warn("Unable to parse max_delay in the Hydrator Mutator, falling pack to default.")
+			} else {
+				maxRetryDelay = d
+			}
 		}
-		switch res.StatusCode {
-		case http.StatusOK:
-			return nil
-		case http.StatusUnauthorized:
-			if cfg.Api.Auth != nil {
-				return errors.New(ErrInvalidCredentials)
+		if len(cfg.Api.Retry.GiveUpAfter) > 0 {
+			if d, err := time.ParseDuration(cfg.Api.Retry.GiveUpAfter); err != nil {
+				a.d.Logger().WithError(err).Warn("Unable to parse max_delay in the Hydrator Mutator, falling pack to default.")
 			} else {
-				return errors.New(ErrNoCredentialsProvided)
+				giveUpAfter = d
 			}
-		default:
-			return errors.New(ErrNon200ResponseFromAPI)
 		}
-	}, backoff.WithMaxRetries(backoff.NewConstantBackOff(time.Millisecond*time.Duration(retryConfig.DelayInMilliseconds)), uint64(retryConfig.NumberOfRetries)))
+
+		client.Transport = httpx.NewResilientRoundTripper(a.client.Transport, maxRetryDelay, giveUpAfter)
+	}
+
+	res, err := client.Do(req)
 	if err != nil {
-		return err
+		return errors.WithStack(err)
+	}
+	defer res.Body.Close()
+
+	switch res.StatusCode {
+	case http.StatusOK:
+	case http.StatusUnauthorized:
+		if cfg.Api.Auth != nil {
+			return errors.New(ErrInvalidCredentials)
+		} else {
+			return errors.New(ErrNoCredentialsProvided)
+		}
+	default:
+		return errors.New(ErrNon200ResponseFromAPI)
 	}
 
 	sessionFromUpstream := authn.AuthenticationSession{}

pipeline/mutate/mutator_hydrator_test.go

@@ -112,9 +112,9 @@ func configWithBasicAuthnForMutator(user, password string) func(*httptest.Server
 	}
 }
 
-func configWithRetriesForMutator(numberOfRetries, retriesDelayInMilliseconds int) func(*httptest.Server) json.RawMessage {
+func configWithRetriesForMutator(giveUpAfter, retryDelay string) func(*httptest.Server) json.RawMessage {
 	return func(s *httptest.Server) json.RawMessage {
-		return []byte(fmt.Sprintf(`{"api": {"url": "%s", "retry": {"number_of_retries": %d, "delay_in_milliseconds": %d}}}`, s.URL, numberOfRetries, retriesDelayInMilliseconds))
+		return []byte(fmt.Sprintf(`{"api": {"url": "%s", "retry": {"give_up_after": "%s", "max_delay": "%s"}}}`, s.URL, giveUpAfter, retryDelay))
 	}
 }
 
@@ -310,7 +310,7 @@ func TestMutatorHydrator(t *testing.T) {
 				Setup:   withInitialErrors(defaultRouterSetup(setExtra(sampleKey, sampleValue)), 2, http.StatusInternalServerError),
 				Session: newAuthenticationSession(),
 				Rule:    &rule.Rule{ID: "test-rule"},
-				Config:  configWithRetriesForMutator(3, 100),
+				Config:  configWithRetriesForMutator("1s", "100ms"),
 				Request: &http.Request{},
 				Match:   newAuthenticationSession(setExtra(sampleKey, sampleValue)),
 				Err:     nil,

rule/rule_migrator.go

@@ -1,6 +1,7 @@
 package rule
 
 import (
+	"fmt"
 	"strings"
 
 	"github.com/blang/semver"
@@ -38,7 +39,51 @@ func migrateRuleJSON(raw []byte) ([]byte, error) {
 		return nil, errors.WithStack(err)
 	}
 
-	if semver.MustParseRange(">=0.19.0-alpha.0")(version) {
+	if semver.MustParseRange("<=0.32.0-beta.1")(version) {
+		// Applies the following patch:
+		//
+		// - number_of_retries (int) => give_up_after (duration, string, := number_of_retries*delay_in_milliseconds + "ms")
+		// - delay_in_milliseconds (int) => max_delay (duration, string, := delay_in_milliseconds + "ms")
+		if mutators := gjson.GetBytes(raw, `mutators`); mutators.Exists() {
+			for key, value := range mutators.Array() {
+				if value.Get("handler").String() != "hydrator" {
+					continue
+				}
+
+				rj := gjson.GetBytes(raw, fmt.Sprintf(`mutators.%d.config.retry.number_of_retries`, key))
+				dj := gjson.GetBytes(raw, fmt.Sprintf(`mutators.%d.config.retry.delay_in_milliseconds`, key))
+
+				var delay = int64(100)
+				var retries = int64(3)
+				var err error
+				if dj.Exists() {
+					delay = dj.Int()
+					if raw, err = sjson.SetBytes(raw, fmt.Sprintf(`mutators.%d.config.retry.max_delay`, key), fmt.Sprintf("%dms", delay)); err != nil {
+						return nil, errors.WithStack(err)
+					}
+
+					if raw, err = sjson.DeleteBytes(raw, fmt.Sprintf(`mutators.%d.config.retry.delay_in_milliseconds`, key)); err != nil {
+						return nil, errors.WithStack(err)
+					}
+				}
+
+				if rj.Exists() {
+					retries = rj.Int()
+					if raw, err = sjson.SetBytes(raw, fmt.Sprintf(`mutators.%d.config.retry.give_up_after`, key), fmt.Sprintf("%dms", retries*delay)); err != nil {
+						return nil, errors.WithStack(err)
+					}
+
+					if raw, err = sjson.DeleteBytes(raw, fmt.Sprintf(`mutators.%d.config.retry.number_of_retries`, key)); err != nil {
+						return nil, errors.WithStack(err)
+					}
+				}
+			}
+		}
+
+		return raw, nil
+	}
+
+	if semver.MustParseRange(">0.32.0-beta.1")(version) {
 		return raw, nil
 	}
 

rule/rule_migrator_test.go

@@ -44,6 +44,43 @@ func TestRuleMigration(t *testing.T) {
 			out:     `{"id":"","version":"v0.19.0-beta.1","description":"","match":{"methods":null,"url":""},"authenticators":null,"authorizer":{"handler":"","config":null},"mutators":null,"upstream":{"preserve_host":false,"strip_path":"","url":""}}`,
 			version: "v0.19.0-beta.1+oryOS.12",
 		},
+		{
+			d: "should migrate to 0.33.0",
+			in: `{
+  "version": "v0.30.0-beta.1",
+  "mutators": [
+	{},	
+    {
+      "handler": "hydrator",
+      "config": {
+        "retry": {
+          "delay_in_milliseconds": 500,
+          "number_of_retries": 5
+        }
+      }
+    }
+  ]
+}`,
+			out: `{
+  "id": "",
+  "version": "v0.33.0-beta.1",
+  "description":"","match":{"methods":null,"url":""},"authenticators":null,"authorizer":{"handler":"","config":null},
+  "mutators": [
+	{"handler":"","config":null},
+    {
+      "handler": "hydrator",
+      "config": {
+        "retry": {
+          "max_delay": "500ms",
+          "give_up_after": "2500ms"
+        }
+      }
+    }
+  ],
+  "upstream":{"preserve_host":false,"strip_path":"","url":""}
+}`,
+			version: "v0.33.0-beta.1+oryOS.12",
+		},
 	} {
 		t.Run(fmt.Sprintf("case=%d/description=%s", k, tc.d), func(t *testing.T) {
 			var r Rule

x/registry.go

@@ -6,6 +6,12 @@ import (
 	"github.com/ory/herodot"
 )
 
+type TestLoggerProvider struct{}
+
+func (lp *TestLoggerProvider) Logger() logrus.FieldLogger {
+	return logrus.New()
+}
+
 type RegistryLogger interface {
 	Logger() logrus.FieldLogger
 }