Fix certmanager provisioning retry and metrics
Address several provisioning defects that either stalled the retry budget or crashed the process: - Classify CAA resolver/transport failures apart from a real CAA policy denial. Both shared the "caa records" wording, so a transient resolver error was persisted as customer misconfiguration and retried forever without consuming the retry budget. A new ErrCAANotPermitted sentinel now marks the genuine misconfiguration; other CAA errors are treated as ordinary transient failures. - Honor an explicit Retry-After: 0 (or a past date) as permission for an immediate retry instead of promoting it to the one-hour default cooldown. acme.RateLimit collapses zero, invalid, and absent headers to a zero duration, so the header is now parsed directly to tell an explicit zero apart from a missing one. - Reuse already-registered Prometheus collectors when a second ACMEService shares a registerer. The fixed-name collectors were MustRegistered, so a duplicate registration panicked the process. - Persist provisioning failures on a context detached from the process tick deadline. A timed-out attempt reached persistFailure with an expired context, so the write-back failed and the retry budget never advanced, leaving the certificate indefinitely retriable. - Use pgx.StrictNamedArgs in the certificate FOR UPDATE loaders to match the coredata SQL contract. Signed-off-by: Bryan Frimin <bryan@probo.com>
This commit is contained in:
@@ -23,6 +23,8 @@ package certmanager
|
|||||||
import (
|
import (
|
||||||
"errors"
|
"errors"
|
||||||
"fmt"
|
"fmt"
|
||||||
|
"net/http"
|
||||||
|
"strconv"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
"golang.org/x/crypto/acme"
|
"golang.org/x/crypto/acme"
|
||||||
@@ -37,12 +39,13 @@ var (
|
|||||||
)
|
)
|
||||||
|
|
||||||
type ACMEError struct {
|
type ACMEError struct {
|
||||||
op string
|
op string
|
||||||
err error
|
err error
|
||||||
problemType string
|
problemType string
|
||||||
detail string
|
detail string
|
||||||
rateLimited bool
|
rateLimited bool
|
||||||
retryAfter time.Duration
|
retryAfter time.Duration
|
||||||
|
retryAfterSet bool
|
||||||
}
|
}
|
||||||
|
|
||||||
func (e *ACMEError) Error() string {
|
func (e *ACMEError) Error() string {
|
||||||
@@ -70,14 +73,20 @@ func (e *ACMEError) Is(target error) bool {
|
|||||||
}
|
}
|
||||||
|
|
||||||
// RetryAfter returns how long callers should wait before retrying.
|
// RetryAfter returns how long callers should wait before retrying.
|
||||||
// For rate-limited errors it prefers the ACME Retry-After value and falls back
|
// For rate-limited errors it honors the ACME Retry-After value whenever the
|
||||||
// to defaultCooldown when the header is absent. Non-rate-limited errors return 0.
|
// header is present and parseable — including a zero (or past) value, which the
|
||||||
|
// CA uses to permit an immediate retry. It falls back to defaultCooldown only
|
||||||
|
// when the header is absent or invalid. Non-rate-limited errors return 0.
|
||||||
func (e *ACMEError) RetryAfter() time.Duration {
|
func (e *ACMEError) RetryAfter() time.Duration {
|
||||||
if e == nil || !e.rateLimited {
|
if e == nil || !e.rateLimited {
|
||||||
return 0
|
return 0
|
||||||
}
|
}
|
||||||
|
|
||||||
if e.retryAfter > 0 {
|
if e.retryAfterSet {
|
||||||
|
if e.retryAfter < 0 {
|
||||||
|
return 0
|
||||||
|
}
|
||||||
|
|
||||||
return e.retryAfter
|
return e.retryAfter
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -119,10 +128,43 @@ func newACMEError(op string, err error) *ACMEError {
|
|||||||
out.problemType = acmeErr.ProblemType
|
out.problemType = acmeErr.ProblemType
|
||||||
out.detail = acmeErr.Detail
|
out.detail = acmeErr.Detail
|
||||||
|
|
||||||
if retryAfter, ok := acme.RateLimit(acmeErr); ok {
|
if _, ok := acme.RateLimit(acmeErr); ok {
|
||||||
out.rateLimited = true
|
out.rateLimited = true
|
||||||
out.retryAfter = retryAfter
|
|
||||||
|
// acme.RateLimit collapses "Retry-After: 0", an invalid header, and a
|
||||||
|
// missing header all to a zero duration, so inspect the header directly
|
||||||
|
// to tell an explicit zero (immediate retry) apart from an absent one
|
||||||
|
// (fall back to defaultCooldown in RetryAfter).
|
||||||
|
if retryAfter, ok := parseRetryAfter(acmeErr.Header); ok {
|
||||||
|
out.retryAfter = retryAfter
|
||||||
|
out.retryAfterSet = true
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
return out
|
return out
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// parseRetryAfter reports the Retry-After delay and whether the header was
|
||||||
|
// present and parseable. It mirrors the delta-seconds and HTTP-date forms the
|
||||||
|
// ACME client understands. An absent or unparseable value returns ok=false so
|
||||||
|
// callers can apply their own fallback.
|
||||||
|
func parseRetryAfter(header http.Header) (time.Duration, bool) {
|
||||||
|
if header == nil {
|
||||||
|
return 0, false
|
||||||
|
}
|
||||||
|
|
||||||
|
value := header.Get("Retry-After")
|
||||||
|
if value == "" {
|
||||||
|
return 0, false
|
||||||
|
}
|
||||||
|
|
||||||
|
if seconds, err := strconv.Atoi(value); err == nil {
|
||||||
|
return time.Duration(seconds) * time.Second, true
|
||||||
|
}
|
||||||
|
|
||||||
|
if date, err := http.ParseTime(value); err == nil {
|
||||||
|
return time.Until(date), true
|
||||||
|
}
|
||||||
|
|
||||||
|
return 0, false
|
||||||
|
}
|
||||||
|
|||||||
@@ -26,11 +26,34 @@ import (
|
|||||||
"testing"
|
"testing"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
|
"github.com/prometheus/client_golang/prometheus"
|
||||||
"github.com/stretchr/testify/assert"
|
"github.com/stretchr/testify/assert"
|
||||||
"github.com/stretchr/testify/require"
|
"github.com/stretchr/testify/require"
|
||||||
"golang.org/x/crypto/acme"
|
"golang.org/x/crypto/acme"
|
||||||
)
|
)
|
||||||
|
|
||||||
|
func TestNewMetrics_SharedRegistererDoesNotPanic(t *testing.T) {
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
registerer := prometheus.NewRegistry()
|
||||||
|
|
||||||
|
first := newMetrics(registerer)
|
||||||
|
require.NotNil(t, first)
|
||||||
|
|
||||||
|
// A second ACMEService sharing the registerer re-registers fixed-name
|
||||||
|
// collectors; this must reuse the existing ones instead of panicking.
|
||||||
|
var second *metrics
|
||||||
|
require.NotPanics(t, func() {
|
||||||
|
second = newMetrics(registerer)
|
||||||
|
})
|
||||||
|
require.NotNil(t, second)
|
||||||
|
|
||||||
|
assert.Same(t, first.provisionSteps, second.provisionSteps)
|
||||||
|
assert.Same(t, first.acmeErrors, second.acmeErrors)
|
||||||
|
assert.Same(t, first.stepDuration, second.stepDuration)
|
||||||
|
assert.Equal(t, first.acmeCooldown, second.acmeCooldown)
|
||||||
|
}
|
||||||
|
|
||||||
func TestNewACMEError_RateLimited(t *testing.T) {
|
func TestNewACMEError_RateLimited(t *testing.T) {
|
||||||
t.Parallel()
|
t.Parallel()
|
||||||
|
|
||||||
@@ -63,6 +86,41 @@ func TestNewACMEError_RateLimitedDefaultCooldown(t *testing.T) {
|
|||||||
assert.Equal(t, defaultCooldown, err.RetryAfter())
|
assert.Equal(t, defaultCooldown, err.RetryAfter())
|
||||||
}
|
}
|
||||||
|
|
||||||
|
func TestNewACMEError_RateLimitedRetryAfterZero(t *testing.T) {
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
err := newACMEError(
|
||||||
|
"cannot create order",
|
||||||
|
&acme.Error{
|
||||||
|
ProblemType: "urn:ietf:params:acme:error:rateLimited",
|
||||||
|
Header: http.Header{"Retry-After": []string{"0"}},
|
||||||
|
},
|
||||||
|
)
|
||||||
|
|
||||||
|
require.NotNil(t, err)
|
||||||
|
assert.ErrorIs(t, err, ErrACMERateLimited)
|
||||||
|
// An explicit Retry-After: 0 permits an immediate retry and must not be
|
||||||
|
// promoted to the one-hour default cooldown.
|
||||||
|
assert.Equal(t, time.Duration(0), err.RetryAfter())
|
||||||
|
}
|
||||||
|
|
||||||
|
func TestNewACMEError_RateLimitedRetryAfterInvalid(t *testing.T) {
|
||||||
|
t.Parallel()
|
||||||
|
|
||||||
|
err := newACMEError(
|
||||||
|
"cannot create order",
|
||||||
|
&acme.Error{
|
||||||
|
ProblemType: "urn:ietf:params:acme:error:rateLimited",
|
||||||
|
Header: http.Header{"Retry-After": []string{"not-a-date"}},
|
||||||
|
},
|
||||||
|
)
|
||||||
|
|
||||||
|
require.NotNil(t, err)
|
||||||
|
assert.ErrorIs(t, err, ErrACMERateLimited)
|
||||||
|
// An unparseable header falls back to the default cooldown.
|
||||||
|
assert.Equal(t, defaultCooldown, err.RetryAfter())
|
||||||
|
}
|
||||||
|
|
||||||
func TestNewACMEError_NonRateLimited(t *testing.T) {
|
func TestNewACMEError_NonRateLimited(t *testing.T) {
|
||||||
t.Parallel()
|
t.Parallel()
|
||||||
|
|
||||||
|
|||||||
@@ -21,6 +21,7 @@
|
|||||||
package certmanager
|
package certmanager
|
||||||
|
|
||||||
import (
|
import (
|
||||||
|
"errors"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
"github.com/prometheus/client_golang/prometheus"
|
"github.com/prometheus/client_golang/prometheus"
|
||||||
@@ -56,48 +57,68 @@ func newMetrics(registerer prometheus.Registerer) *metrics {
|
|||||||
registerer = prometheus.DefaultRegisterer
|
registerer = prometheus.DefaultRegisterer
|
||||||
}
|
}
|
||||||
|
|
||||||
m := &metrics{
|
return &metrics{
|
||||||
provisionSteps: prometheus.NewCounterVec(
|
provisionSteps: registerCollector(
|
||||||
prometheus.CounterOpts{
|
registerer,
|
||||||
Subsystem: "certmanager",
|
prometheus.NewCounterVec(
|
||||||
Name: "certificate_provision_steps_total",
|
prometheus.CounterOpts{
|
||||||
Help: "Certificate provisioning steps by phase and result.",
|
Subsystem: "certmanager",
|
||||||
},
|
Name: "certificate_provision_steps_total",
|
||||||
[]string{"phase", "result"},
|
Help: "Certificate provisioning steps by phase and result.",
|
||||||
|
},
|
||||||
|
[]string{"phase", "result"},
|
||||||
|
),
|
||||||
),
|
),
|
||||||
acmeErrors: prometheus.NewCounterVec(
|
acmeErrors: registerCollector(
|
||||||
prometheus.CounterOpts{
|
registerer,
|
||||||
Subsystem: "certmanager",
|
prometheus.NewCounterVec(
|
||||||
Name: "certificate_acme_errors_total",
|
prometheus.CounterOpts{
|
||||||
Help: "ACME errors by problem type.",
|
Subsystem: "certmanager",
|
||||||
},
|
Name: "certificate_acme_errors_total",
|
||||||
[]string{"problem_type"},
|
Help: "ACME errors by problem type.",
|
||||||
|
},
|
||||||
|
[]string{"problem_type"},
|
||||||
|
),
|
||||||
),
|
),
|
||||||
acmeCooldown: prometheus.NewGauge(
|
acmeCooldown: registerCollector(
|
||||||
prometheus.GaugeOpts{
|
registerer,
|
||||||
Subsystem: "certmanager",
|
prometheus.NewGauge(
|
||||||
Name: "certificate_acme_cooldown",
|
prometheus.GaugeOpts{
|
||||||
Help: "1 while the ACME client is in a global rate-limit cooldown.",
|
Subsystem: "certmanager",
|
||||||
},
|
Name: "certificate_acme_cooldown",
|
||||||
|
Help: "1 while the ACME client is in a global rate-limit cooldown.",
|
||||||
|
},
|
||||||
|
),
|
||||||
),
|
),
|
||||||
stepDuration: prometheus.NewHistogramVec(
|
stepDuration: registerCollector(
|
||||||
prometheus.HistogramOpts{
|
registerer,
|
||||||
Subsystem: "certmanager",
|
prometheus.NewHistogramVec(
|
||||||
Name: "certificate_provision_step_duration_seconds",
|
prometheus.HistogramOpts{
|
||||||
Help: "Duration of certificate provisioning steps in seconds.",
|
Subsystem: "certmanager",
|
||||||
},
|
Name: "certificate_provision_step_duration_seconds",
|
||||||
[]string{"phase"},
|
Help: "Duration of certificate provisioning steps in seconds.",
|
||||||
|
},
|
||||||
|
[]string{"phase"},
|
||||||
|
),
|
||||||
),
|
),
|
||||||
}
|
}
|
||||||
|
}
|
||||||
|
|
||||||
registerer.MustRegister(
|
func registerCollector[T prometheus.Collector](
|
||||||
m.provisionSteps,
|
registerer prometheus.Registerer,
|
||||||
m.acmeErrors,
|
collector T,
|
||||||
m.acmeCooldown,
|
) T {
|
||||||
m.stepDuration,
|
if err := registerer.Register(collector); err != nil {
|
||||||
)
|
if already, ok := errors.AsType[prometheus.AlreadyRegisteredError](err); ok {
|
||||||
|
if existing, ok := already.ExistingCollector.(T); ok {
|
||||||
|
return existing
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
return m
|
panic(err)
|
||||||
|
}
|
||||||
|
|
||||||
|
return collector
|
||||||
}
|
}
|
||||||
|
|
||||||
func (m *metrics) observeStep(phase provisionPhase, result provisionResult, started time.Time) {
|
func (m *metrics) observeStep(phase provisionPhase, result provisionResult, started time.Time) {
|
||||||
|
|||||||
@@ -43,6 +43,10 @@ const (
|
|||||||
maxProvisioningRetries = 3
|
maxProvisioningRetries = 3
|
||||||
dnsExchangeTimeout = 10 * time.Second
|
dnsExchangeTimeout = 10 * time.Second
|
||||||
processTickTimeout = 90 * time.Second
|
processTickTimeout = 90 * time.Second
|
||||||
|
// persistFailureTimeout bounds the retry-outcome write-back. It runs on a
|
||||||
|
// context detached from the process tick deadline so a timed-out attempt can
|
||||||
|
// still record its failure.
|
||||||
|
persistFailureTimeout = 15 * time.Second
|
||||||
|
|
||||||
tracerName = "go.probo.inc/probo/pkg/certmanager"
|
tracerName = "go.probo.inc/probo/pkg/certmanager"
|
||||||
)
|
)
|
||||||
@@ -404,6 +408,15 @@ func (h *provisionHandler) persistFailure(
|
|||||||
) error {
|
) error {
|
||||||
errorCode := classifyProvisioningError(provisionErr)
|
errorCode := classifyProvisioningError(provisionErr)
|
||||||
|
|
||||||
|
// Process runs each tick under processTickTimeout. When that deadline fires
|
||||||
|
// mid-attempt, the same expired context reaches here, and the write-back
|
||||||
|
// silently fails — so the retry budget never advances and the certificate
|
||||||
|
// stays retriable forever. Detach from the tick deadline (and cancellation)
|
||||||
|
// and bound the write with its own timeout so repeated timeouts still make
|
||||||
|
// progress toward FAILED.
|
||||||
|
ctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), persistFailureTimeout)
|
||||||
|
defer cancel()
|
||||||
|
|
||||||
return h.pg.WithTx(
|
return h.pg.WithTx(
|
||||||
ctx,
|
ctx,
|
||||||
func(ctx context.Context, tx pg.Tx) error {
|
func(ctx context.Context, tx pg.Tx) error {
|
||||||
@@ -731,7 +744,8 @@ func (h *provisionHandler) checkCAARecords(ctx context.Context, hostname string)
|
|||||||
}
|
}
|
||||||
|
|
||||||
return fmt.Errorf(
|
return fmt.Errorf(
|
||||||
"caa records for domain %q do not permit issuance by %q",
|
"%w: domain %q by %q",
|
||||||
|
ErrCAANotPermitted,
|
||||||
hostname,
|
hostname,
|
||||||
h.caaIssuerDomain,
|
h.caaIssuerDomain,
|
||||||
)
|
)
|
||||||
|
|||||||
@@ -22,6 +22,7 @@ package certmanager
|
|||||||
|
|
||||||
import (
|
import (
|
||||||
"errors"
|
"errors"
|
||||||
|
"fmt"
|
||||||
"testing"
|
"testing"
|
||||||
|
|
||||||
"github.com/stretchr/testify/assert"
|
"github.com/stretchr/testify/assert"
|
||||||
@@ -35,8 +36,11 @@ func TestClassifyProvisioningError(t *testing.T) {
|
|||||||
assert.Equal(t, ProvisioningErrorACMERateLimited, classifyProvisioningError(ErrACMERateLimited))
|
assert.Equal(t, ProvisioningErrorACMERateLimited, classifyProvisioningError(ErrACMERateLimited))
|
||||||
assert.Equal(t, ProvisioningErrorACMEInvalidOrder, classifyProvisioningError(ErrOrderInvalid))
|
assert.Equal(t, ProvisioningErrorACMEInvalidOrder, classifyProvisioningError(ErrOrderInvalid))
|
||||||
assert.Equal(t, ProvisioningErrorDNSCNAME, classifyProvisioningError(errors.New("cname target mismatch")))
|
assert.Equal(t, ProvisioningErrorDNSCNAME, classifyProvisioningError(errors.New("cname target mismatch")))
|
||||||
assert.Equal(t, ProvisioningErrorDNSCAA, classifyProvisioningError(errors.New("caa records for domain")))
|
assert.Equal(t, ProvisioningErrorDNSCAA, classifyProvisioningError(fmt.Errorf("%w: domain %q", ErrCAANotPermitted, "example.com")))
|
||||||
assert.Equal(t, ProvisioningErrorACMETemporary, classifyProvisioningError(errors.New("network timeout")))
|
assert.Equal(t, ProvisioningErrorACMETemporary, classifyProvisioningError(errors.New("network timeout")))
|
||||||
|
// A CAA resolver/transport failure shares the "caa records" wording with a
|
||||||
|
// real CAA misconfiguration but must consume the normal retry budget.
|
||||||
|
assert.Equal(t, ProvisioningErrorACMETemporary, classifyProvisioningError(errors.New("cannot exchange dns message for caa records: i/o timeout")))
|
||||||
}
|
}
|
||||||
|
|
||||||
func TestDecideProvisioningOutcome_RateLimitKeepsRetryCountAndOrder(t *testing.T) {
|
func TestDecideProvisioningOutcome_RateLimitKeepsRetryCountAndOrder(t *testing.T) {
|
||||||
|
|||||||
@@ -34,6 +34,13 @@ const (
|
|||||||
ProvisioningErrorACMEFailed = "ACME_FAILED"
|
ProvisioningErrorACMEFailed = "ACME_FAILED"
|
||||||
)
|
)
|
||||||
|
|
||||||
|
// ErrCAANotPermitted marks a CAA policy that actively forbids issuance by our
|
||||||
|
// CA. This is a customer-side misconfiguration and is intentionally
|
||||||
|
// non-terminal. It must stay distinct from a CAA resolver/transport failure,
|
||||||
|
// which shares the "caa records" wording but is a transient error that has to
|
||||||
|
// consume the normal retry budget instead of retrying forever.
|
||||||
|
var ErrCAANotPermitted = errors.New("caa records do not permit issuance")
|
||||||
|
|
||||||
func classifyProvisioningError(err error) string {
|
func classifyProvisioningError(err error) string {
|
||||||
if err == nil {
|
if err == nil {
|
||||||
return ""
|
return ""
|
||||||
@@ -47,12 +54,14 @@ func classifyProvisioningError(err error) string {
|
|||||||
return ProvisioningErrorACMEInvalidOrder
|
return ProvisioningErrorACMEInvalidOrder
|
||||||
}
|
}
|
||||||
|
|
||||||
|
if errors.Is(err, ErrCAANotPermitted) {
|
||||||
|
return ProvisioningErrorDNSCAA
|
||||||
|
}
|
||||||
|
|
||||||
msg := strings.ToLower(err.Error())
|
msg := strings.ToLower(err.Error())
|
||||||
switch {
|
switch {
|
||||||
case strings.Contains(msg, "cname"):
|
case strings.Contains(msg, "cname"):
|
||||||
return ProvisioningErrorDNSCNAME
|
return ProvisioningErrorDNSCNAME
|
||||||
case strings.Contains(msg, "caa record"):
|
|
||||||
return ProvisioningErrorDNSCAA
|
|
||||||
case strings.Contains(msg, "status: invalid"), strings.Contains(msg, "order is in unexpected status \"invalid\""):
|
case strings.Contains(msg, "status: invalid"), strings.Contains(msg, "order is in unexpected status \"invalid\""):
|
||||||
return ProvisioningErrorACMEInvalidOrder
|
return ProvisioningErrorACMEInvalidOrder
|
||||||
default:
|
default:
|
||||||
|
|||||||
@@ -220,7 +220,7 @@ FOR UPDATE SKIP LOCKED
|
|||||||
|
|
||||||
q = fmt.Sprintf(q, scope.SQLFragment())
|
q = fmt.Sprintf(q, scope.SQLFragment())
|
||||||
|
|
||||||
args := pgx.NamedArgs{"id": certificateID}
|
args := pgx.StrictNamedArgs{"id": certificateID}
|
||||||
maps.Copy(args, scope.SQLArguments())
|
maps.Copy(args, scope.SQLArguments())
|
||||||
|
|
||||||
rows, err := conn.Query(ctx, q, args)
|
rows, err := conn.Query(ctx, q, args)
|
||||||
@@ -280,7 +280,7 @@ FOR UPDATE
|
|||||||
|
|
||||||
q = fmt.Sprintf(q, scope.SQLFragment())
|
q = fmt.Sprintf(q, scope.SQLFragment())
|
||||||
|
|
||||||
args := pgx.NamedArgs{"id": certificateID}
|
args := pgx.StrictNamedArgs{"id": certificateID}
|
||||||
maps.Copy(args, scope.SQLArguments())
|
maps.Copy(args, scope.SQLArguments())
|
||||||
|
|
||||||
rows, err := conn.Query(ctx, q, args)
|
rows, err := conn.Query(ctx, q, args)
|
||||||
|
|||||||
Reference in New Issue
Block a user