From 1f91cffa200502f74890f5436af46e48c92ef434 Mon Sep 17 00:00:00 2001 From: Matteo Gastaldello Date: Tue, 14 Apr 2026 14:40:02 +0200 Subject: [PATCH 1/4] feat(metrics): add telemetry metrics for reconciliation processes - Introduced a MetricsRecorder interface to define metrics recording methods. - Implemented metrics recording in the Reconciler for various operations including Get, Connect, Observe, Create, Update, Delete, Finalizer Add/Remove, and Status Update. - Added a new telemetry package with OpenTelemetry integration for metrics collection. - Created a Metrics struct to encapsulate all metrics related to the provider-runtime. - Implemented Setup function to initialize OpenTelemetry metrics pipeline. - Added unit tests to verify metrics recording functionality. --- go.mod | 32 ++- go.sum | 91 +++--- pkg/controller/options.go | 24 +- pkg/controller/queue_wait.go | 193 +++++++++++++ pkg/controller/queue_wait_test.go | 171 +++++++++++ pkg/reconciler/reconciler.go | 287 +++++++++++++++++-- pkg/telemetry/metrics.go | 455 ++++++++++++++++++++++++++++++ pkg/telemetry/metrics_test.go | 117 ++++++++ 8 files changed, 1295 insertions(+), 75 deletions(-) create mode 100644 pkg/controller/queue_wait.go create mode 100644 pkg/controller/queue_wait_test.go create mode 100644 pkg/telemetry/metrics.go create mode 100644 pkg/telemetry/metrics_test.go diff --git a/go.mod b/go.mod index 39ca32f..73354a2 100644 --- a/go.mod +++ b/go.mod @@ -6,6 +6,11 @@ require ( github.com/go-logr/logr v1.4.3 github.com/google/go-cmp v0.7.0 github.com/krateoplatformops/plumbing v1.7.1 + go.opentelemetry.io/otel v1.43.0 + go.opentelemetry.io/otel/exporters/otlp/otlpmetric/otlpmetrichttp v1.43.0 + go.opentelemetry.io/otel/metric v1.43.0 + go.opentelemetry.io/otel/sdk v1.43.0 + go.opentelemetry.io/otel/sdk/metric v1.43.0 golang.org/x/time v0.12.0 k8s.io/api v0.35.3 k8s.io/apimachinery v0.35.3 @@ -16,6 +21,7 @@ require ( require ( github.com/beorn7/perks v1.0.1 // indirect + github.com/cenkalti/backoff/v5 v5.0.3 // indirect github.com/cespare/xxhash/v2 v2.3.0 // indirect github.com/davecgh/go-spew v1.1.2-0.20180830191138-d8f796af33cc // indirect github.com/emicklei/go-restful/v3 v3.12.2 // indirect @@ -23,6 +29,7 @@ require ( github.com/fatih/color v1.18.0 // indirect github.com/fsnotify/fsnotify v1.9.0 // indirect github.com/fxamacker/cbor/v2 v2.9.0 // indirect + github.com/go-logr/stdr v1.2.2 // indirect github.com/go-openapi/jsonpointer v0.21.1 // indirect github.com/go-openapi/jsonreference v0.21.0 // indirect github.com/go-openapi/swag v0.23.1 // indirect @@ -30,6 +37,7 @@ require ( github.com/google/btree v1.1.3 // indirect github.com/google/gnostic-models v0.7.0 // indirect github.com/google/uuid v1.6.0 // indirect + github.com/grpc-ecosystem/grpc-gateway/v2 v2.28.0 // indirect github.com/inconshreveable/mousetrap v1.1.0 // indirect github.com/josharian/intern v1.0.0 // indirect github.com/json-iterator/go v1.1.12 // indirect @@ -47,18 +55,24 @@ require ( github.com/spf13/cobra v1.10.2 // indirect github.com/spf13/pflag v1.0.10 // indirect github.com/x448/float16 v0.8.4 // indirect + go.opentelemetry.io/auto/sdk v1.2.1 // indirect + go.opentelemetry.io/otel/trace v1.43.0 // indirect + go.opentelemetry.io/proto/otlp v1.10.0 // indirect go.yaml.in/yaml/v2 v2.4.3 // indirect go.yaml.in/yaml/v3 v3.0.4 // indirect - golang.org/x/mod v0.31.0 // indirect - golang.org/x/net v0.48.0 // indirect - golang.org/x/oauth2 v0.30.0 // indirect - golang.org/x/sync v0.19.0 // indirect - golang.org/x/sys v0.40.0 // indirect - golang.org/x/term v0.39.0 // indirect - golang.org/x/text v0.33.0 // indirect - golang.org/x/tools v0.40.0 // indirect + golang.org/x/mod v0.33.0 // indirect + golang.org/x/net v0.52.0 // indirect + golang.org/x/oauth2 v0.35.0 // indirect + golang.org/x/sync v0.20.0 // indirect + golang.org/x/sys v0.42.0 // indirect + golang.org/x/term v0.41.0 // indirect + golang.org/x/text v0.35.0 // indirect + golang.org/x/tools v0.42.0 // indirect gomodules.xyz/jsonpatch/v2 v2.4.0 // indirect - google.golang.org/protobuf v1.36.8 // indirect + google.golang.org/genproto/googleapis/api v0.0.0-20260401024825-9d38bb4040a9 // indirect + google.golang.org/genproto/googleapis/rpc v0.0.0-20260401024825-9d38bb4040a9 // indirect + google.golang.org/grpc v1.80.0 // indirect + google.golang.org/protobuf v1.36.11 // indirect gopkg.in/evanphx/json-patch.v4 v4.13.0 // indirect gopkg.in/inf.v0 v0.9.1 // indirect gopkg.in/yaml.v2 v2.4.0 // indirect diff --git a/go.sum b/go.sum index 0d38e08..19ca6de 100644 --- a/go.sum +++ b/go.sum @@ -1,5 +1,5 @@ -cel.dev/expr v0.24.0 h1:56OvJKSH3hDGL0ml5uSxZmz3/3Pq4tJ+fb1unVLAFcY= -cel.dev/expr v0.24.0/go.mod h1:hLPLo1W4QUmuYdA72RBX06QTs6MXw941piREPl3Yfiw= +cel.dev/expr v0.25.1 h1:1KrZg61W6TWSxuNZ37Xy49ps13NUovb66QLprthtwi4= +cel.dev/expr v0.25.1/go.mod h1:hrXvqGP6G6gyx8UAHSHJ5RGk//1Oj5nXQ2NI02Nrsg4= github.com/Masterminds/semver/v3 v3.4.0 h1:Zog+i5UMtVoCU8oKka5P7i9q9HgrJeGzI9SA1Xbatp0= github.com/Masterminds/semver/v3 v3.4.0/go.mod h1:4V+yj/TJE1HU9XfppCwVMZq3I84lprf4nC11bSS5beM= github.com/antlr4-go/antlr/v4 v4.13.0 h1:lxCg3LAv+EUK6t1i0y1V6/SLeUi0eKEKdhQAlS8TVTI= @@ -10,6 +10,8 @@ github.com/blang/semver/v4 v4.0.0 h1:1PFHFE6yCCTv8C1TeyNNarDzntLi7wMI5i/pzqYIsAM github.com/blang/semver/v4 v4.0.0/go.mod h1:IbckMUScFkM3pff0VJDNKRiT6TG/YpiHIM2yvyW5YoQ= github.com/cenkalti/backoff/v4 v4.3.0 h1:MyRJ/UdXutAwSAT+s3wNd7MfTIcy71VQueUuFK343L8= github.com/cenkalti/backoff/v4 v4.3.0/go.mod h1:Y3VNntkOUPxTVeUxJ/G5vcM//AlwfmyYozVcomhLiZE= +github.com/cenkalti/backoff/v5 v5.0.3 h1:ZN+IMa753KfX5hd8vVaMixjnqRZ3y8CuJKRKj1xcsSM= +github.com/cenkalti/backoff/v5 v5.0.3/go.mod h1:rkhZdG3JZukswDf7f0cwqPNk4K0sa+F97BxZthm/crw= github.com/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs= github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs= github.com/cpuguy83/go-md2man/v2 v2.0.6/go.mod h1:oOW0eioCTA6cOiMLiUPZOpcVxMig6NIQQ7OS05n1F4g= @@ -31,6 +33,7 @@ github.com/fsnotify/fsnotify v1.9.0 h1:2Ml+OJNzbYCTzsxtv8vKSFD9PbJjmhYF14k/jKC7S github.com/fsnotify/fsnotify v1.9.0/go.mod h1:8jBTzvmWwFyi3Pb8djgCCO5IBqzKJ/Jwo8TRcHyHii0= github.com/fxamacker/cbor/v2 v2.9.0 h1:NpKPmjDBgUfBms6tr6JZkTHtfFGcMKsw3eGcmD/sapM= github.com/fxamacker/cbor/v2 v2.9.0/go.mod h1:vM4b+DJCtHn+zz7h3FFp/hDAI9WNWCsZj23V5ytsSxQ= +github.com/go-logr/logr v1.2.2/go.mod h1:jdQByPbusPIv2/zmleS9BjJVeZ6kBagPoEUsqbVz/1A= github.com/go-logr/logr v1.4.3 h1:CjnDlHq8ikf6E492q6eKboGOC0T8CDaOvkHCIg8idEI= github.com/go-logr/logr v1.4.3/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY= github.com/go-logr/stdr v1.2.2 h1:hSWxHoqTgW2S2qGc0LTAI563KZ5YKYRhT3MFKZMbjag= @@ -47,6 +50,8 @@ github.com/go-task/slim-sprig/v3 v3.0.0 h1:sUs3vkvUymDpBKi3qH1YSqBQk9+9D/8M2mN1v github.com/go-task/slim-sprig/v3 v3.0.0/go.mod h1:W848ghGpv3Qj3dhTPRyJypKRiqCdHZiAzKg9hl15HA8= github.com/gobuffalo/flect v1.0.3 h1:xeWBM2nui+qnVvNM4S3foBhCAL2XgPU+a7FdpelbTq4= github.com/gobuffalo/flect v1.0.3/go.mod h1:A5msMlrHtLqh9umBSnvabjsMrCcCpAyzglnDvkbYKHs= +github.com/golang/protobuf v1.5.4 h1:i7eJL8qZTpSEXOPTxNKhASYpMn+8e5Q6AdndVa1dWek= +github.com/golang/protobuf v1.5.4/go.mod h1:lnTiLA8Wa4RWRcIUkrtSVa5nRhsEGBg48fD6rSs7xps= github.com/google/btree v1.1.3 h1:CVpQJjYgC4VbzxeGVHfvZrv1ctoYCAI8vbl07Fcxlyg= github.com/google/btree v1.1.3/go.mod h1:qOPhT0dTNdNzV6Z/lhRX0YXUafgPLFUh+gZMl761Gm4= github.com/google/cel-go v0.26.0 h1:DPGjXackMpJWH680oGY4lZhYjIameYmR+/6RBdDGmaI= @@ -62,8 +67,8 @@ github.com/google/pprof v0.0.0-20250630185457-6e76a2b096b5 h1:xhMrHhTJ6zxu3gA4en github.com/google/pprof v0.0.0-20250630185457-6e76a2b096b5/go.mod h1:5hDyRhoBCxViHszMt12TnOpEI4VVi+U8Gm9iphldiMA= github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0= github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo= -github.com/grpc-ecosystem/grpc-gateway/v2 v2.26.3 h1:5ZPtiqj0JL5oKWmcsq4VMaAW5ukBEgSGXEN89zeH1Jo= -github.com/grpc-ecosystem/grpc-gateway/v2 v2.26.3/go.mod h1:ndYquD05frm2vACXE1nsccT4oJzjhw2arTS2cpUD1PI= +github.com/grpc-ecosystem/grpc-gateway/v2 v2.28.0 h1:HWRh5R2+9EifMyIHV7ZV+MIZqgz+PMpZ14Jynv3O2Zs= +github.com/grpc-ecosystem/grpc-gateway/v2 v2.28.0/go.mod h1:JfhWUomR1baixubs02l85lZYYOm7LV6om4ceouMv45c= github.com/inconshreveable/mousetrap v1.1.0 h1:wN+x4NVGpMsO7ErUn/mUI3vEoE6Jt13X2s0bqwp9tc8= github.com/inconshreveable/mousetrap v1.1.0/go.mod h1:vpF70FUmC8bwa3OWnCshd2FqLfsEA9PFc4w1p2J65bw= github.com/josharian/intern v1.0.0 h1:vlS4z54oSdjm0bgjRigI+G1HpF+tI+9rE5LLzOg8HmY= @@ -137,24 +142,28 @@ github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U= github.com/x448/float16 v0.8.4 h1:qLwI1I70+NjRFUR3zs1JPUCgaCXSh3SW62uAKT1mSBM= github.com/x448/float16 v0.8.4/go.mod h1:14CWIYCyZA/cWjXOioeEpHeN/83MdbZDRQHoFcYsOfg= -go.opentelemetry.io/auto/sdk v1.1.0 h1:cH53jehLUN6UFLY71z+NDOiNJqDdPRaXzTel0sJySYA= -go.opentelemetry.io/auto/sdk v1.1.0/go.mod h1:3wSPjt5PWp2RhlCcmmOial7AvC4DQqZb7a7wCow3W8A= +go.opentelemetry.io/auto/sdk v1.2.1 h1:jXsnJ4Lmnqd11kwkBV2LgLoFMZKizbCi5fNZ/ipaZ64= +go.opentelemetry.io/auto/sdk v1.2.1/go.mod h1:KRTj+aOaElaLi+wW1kO/DZRXwkF4C5xPbEe3ZiIhN7Y= go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.61.0 h1:F7Jx+6hwnZ41NSFTO5q4LYDtJRXBf2PD0rNBkeB/lus= go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp v0.61.0/go.mod h1:UHB22Z8QsdRDrnAtX4PntOl36ajSxcdUMt1sF7Y6E7Q= -go.opentelemetry.io/otel v1.37.0 h1:9zhNfelUvx0KBfu/gb+ZgeAfAgtWrfHJZcAqFC228wQ= -go.opentelemetry.io/otel v1.37.0/go.mod h1:ehE/umFRLnuLa/vSccNq9oS1ErUlkkK71gMcN34UG8I= +go.opentelemetry.io/otel v1.43.0 h1:mYIM03dnh5zfN7HautFE4ieIig9amkNANT+xcVxAj9I= +go.opentelemetry.io/otel v1.43.0/go.mod h1:JuG+u74mvjvcm8vj8pI5XiHy1zDeoCS2LB1spIq7Ay0= +go.opentelemetry.io/otel/exporters/otlp/otlpmetric/otlpmetrichttp v1.43.0 h1:w1K+pCJoPpQifuVpsKamUdn9U0zM3xUziVOqsGksUrY= +go.opentelemetry.io/otel/exporters/otlp/otlpmetric/otlpmetrichttp v1.43.0/go.mod h1:HBy4BjzgVE8139ieRI75oXm3EcDN+6GhD88JT1Kjvxg= go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.34.0 h1:OeNbIYk/2C15ckl7glBlOBp5+WlYsOElzTNmiPW/x60= go.opentelemetry.io/otel/exporters/otlp/otlptrace v1.34.0/go.mod h1:7Bept48yIeqxP2OZ9/AqIpYS94h2or0aB4FypJTc8ZM= go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc v1.34.0 h1:tgJ0uaNS4c98WRNUEx5U3aDlrDOI5Rs+1Vifcw4DJ8U= go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc v1.34.0/go.mod h1:U7HYyW0zt/a9x5J1Kjs+r1f/d4ZHnYFclhYY2+YbeoE= -go.opentelemetry.io/otel/metric v1.36.0 h1:MoWPKVhQvJ+eeXWHFBOPoBOi20jh6Iq2CcCREuTYufE= -go.opentelemetry.io/otel/metric v1.36.0/go.mod h1:zC7Ks+yeyJt4xig9DEw9kuUFe5C3zLbVjV2PzT6qzbs= -go.opentelemetry.io/otel/sdk v1.36.0 h1:b6SYIuLRs88ztox4EyrvRti80uXIFy+Sqzoh9kFULbs= -go.opentelemetry.io/otel/sdk v1.36.0/go.mod h1:+lC+mTgD+MUWfjJubi2vvXWcVxyr9rmlshZni72pXeY= -go.opentelemetry.io/otel/trace v1.37.0 h1:HLdcFNbRQBE2imdSEgm/kwqmQj1Or1l/7bW6mxVK7z4= -go.opentelemetry.io/otel/trace v1.37.0/go.mod h1:TlgrlQ+PtQO5XFerSPUYG0JSgGyryXewPGyayAWSBS0= -go.opentelemetry.io/proto/otlp v1.5.0 h1:xJvq7gMzB31/d406fB8U5CBdyQGw4P399D1aQWU/3i4= -go.opentelemetry.io/proto/otlp v1.5.0/go.mod h1:keN8WnHxOy8PG0rQZjJJ5A2ebUoafqWp0eVQ4yIXvJ4= +go.opentelemetry.io/otel/metric v1.43.0 h1:d7638QeInOnuwOONPp4JAOGfbCEpYb+K6DVWvdxGzgM= +go.opentelemetry.io/otel/metric v1.43.0/go.mod h1:RDnPtIxvqlgO8GRW18W6Z/4P462ldprJtfxHxyKd2PY= +go.opentelemetry.io/otel/sdk v1.43.0 h1:pi5mE86i5rTeLXqoF/hhiBtUNcrAGHLKQdhg4h4V9Dg= +go.opentelemetry.io/otel/sdk v1.43.0/go.mod h1:P+IkVU3iWukmiit/Yf9AWvpyRDlUeBaRg6Y+C58QHzg= +go.opentelemetry.io/otel/sdk/metric v1.43.0 h1:S88dyqXjJkuBNLeMcVPRFXpRw2fuwdvfCGLEo89fDkw= +go.opentelemetry.io/otel/sdk/metric v1.43.0/go.mod h1:C/RJtwSEJ5hzTiUz5pXF1kILHStzb9zFlIEe85bhj6A= +go.opentelemetry.io/otel/trace v1.43.0 h1:BkNrHpup+4k4w+ZZ86CZoHHEkohws8AY+WTX09nk+3A= +go.opentelemetry.io/otel/trace v1.43.0/go.mod h1:/QJhyVBUUswCphDVxq+8mld+AvhXZLhe+8WVFxiFff0= +go.opentelemetry.io/proto/otlp v1.10.0 h1:IQRWgT5srOCYfiWnpqUYz9CVmbO8bFmKcwYxpuCSL2g= +go.opentelemetry.io/proto/otlp v1.10.0/go.mod h1:/CV4QoCR/S9yaPj8utp3lvQPoqMtxXdzn7ozvvozVqk= go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto= go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE= go.uber.org/multierr v1.11.0 h1:blXXJkSxSSfBVBlC76pxqeO+LN3aDfLQo+309xJstO0= @@ -167,40 +176,42 @@ go.yaml.in/yaml/v3 v3.0.4 h1:tfq32ie2Jv2UxXFdLJdh3jXuOzWiL1fo0bu/FbuKpbc= go.yaml.in/yaml/v3 v3.0.4/go.mod h1:DhzuOOF2ATzADvBadXxruRBLzYTpT36CKvDb3+aBEFg= golang.org/x/exp v0.0.0-20240719175910-8a7402abbf56 h1:2dVuKD2vS7b0QIHQbpyTISPd0LeHDbnYEryqj5Q1ug8= golang.org/x/exp v0.0.0-20240719175910-8a7402abbf56/go.mod h1:M4RDyNAINzryxdtnbRXRL/OHtkFuWGRjvuhBJpk2IlY= -golang.org/x/mod v0.31.0 h1:HaW9xtz0+kOcWKwli0ZXy79Ix+UW/vOfmWI5QVd2tgI= -golang.org/x/mod v0.31.0/go.mod h1:43JraMp9cGx1Rx3AqioxrbrhNsLl2l/iNAvuBkrezpg= -golang.org/x/net v0.48.0 h1:zyQRTTrjc33Lhh0fBgT/H3oZq9WuvRR5gPC70xpDiQU= -golang.org/x/net v0.48.0/go.mod h1:+ndRgGjkh8FGtu1w1FGbEC31if4VrNVMuKTgcAAnQRY= -golang.org/x/oauth2 v0.30.0 h1:dnDm7JmhM45NNpd8FDDeLhK6FwqbOf4MLCM9zb1BOHI= -golang.org/x/oauth2 v0.30.0/go.mod h1:B++QgG3ZKulg6sRPGD/mqlHQs5rB3Ml9erfeDY7xKlU= -golang.org/x/sync v0.19.0 h1:vV+1eWNmZ5geRlYjzm2adRgW2/mcpevXNg50YZtPCE4= -golang.org/x/sync v0.19.0/go.mod h1:9KTHXmSnoGruLpwFjVSX0lNNA75CykiMECbovNTZqGI= +golang.org/x/mod v0.33.0 h1:tHFzIWbBifEmbwtGz65eaWyGiGZatSrT9prnU8DbVL8= +golang.org/x/mod v0.33.0/go.mod h1:swjeQEj+6r7fODbD2cqrnje9PnziFuw4bmLbBZFrQ5w= +golang.org/x/net v0.52.0 h1:He/TN1l0e4mmR3QqHMT2Xab3Aj3L9qjbhRm78/6jrW0= +golang.org/x/net v0.52.0/go.mod h1:R1MAz7uMZxVMualyPXb+VaqGSa3LIaUqk0eEt3w36Sw= +golang.org/x/oauth2 v0.35.0 h1:Mv2mzuHuZuY2+bkyWXIHMfhNdJAdwW3FuWeCPYN5GVQ= +golang.org/x/oauth2 v0.35.0/go.mod h1:lzm5WQJQwKZ3nwavOZ3IS5Aulzxi68dUSgRHujetwEA= +golang.org/x/sync v0.20.0 h1:e0PTpb7pjO8GAtTs2dQ6jYa5BWYlMuX047Dco/pItO4= +golang.org/x/sync v0.20.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0= golang.org/x/sys v0.0.0-20220811171246-fbc7d0a398ab/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= golang.org/x/sys v0.6.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg= -golang.org/x/sys v0.40.0 h1:DBZZqJ2Rkml6QMQsZywtnjnnGvHza6BTfYFWY9kjEWQ= -golang.org/x/sys v0.40.0/go.mod h1:OgkHotnGiDImocRcuBABYBEXf8A9a87e/uXjp9XT3ks= -golang.org/x/term v0.39.0 h1:RclSuaJf32jOqZz74CkPA9qFuVTX7vhLlpfj/IGWlqY= -golang.org/x/term v0.39.0/go.mod h1:yxzUCTP/U+FzoxfdKmLaA0RV1WgE0VY7hXBwKtY/4ww= -golang.org/x/text v0.33.0 h1:B3njUFyqtHDUI5jMn1YIr5B0IE2U0qck04r6d4KPAxE= -golang.org/x/text v0.33.0/go.mod h1:LuMebE6+rBincTi9+xWTY8TztLzKHc/9C1uBCG27+q8= +golang.org/x/sys v0.42.0 h1:omrd2nAlyT5ESRdCLYdm3+fMfNFE/+Rf4bDIQImRJeo= +golang.org/x/sys v0.42.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw= +golang.org/x/term v0.41.0 h1:QCgPso/Q3RTJx2Th4bDLqML4W6iJiaXFq2/ftQF13YU= +golang.org/x/term v0.41.0/go.mod h1:3pfBgksrReYfZ5lvYM0kSO0LIkAl4Yl2bXOkKP7Ec2A= +golang.org/x/text v0.35.0 h1:JOVx6vVDFokkpaq1AEptVzLTpDe9KGpj5tR4/X+ybL8= +golang.org/x/text v0.35.0/go.mod h1:khi/HExzZJ2pGnjenulevKNX1W67CUy0AsXcNubPGCA= golang.org/x/time v0.12.0 h1:ScB/8o8olJvc+CQPWrK3fPZNfh7qgwCrY0zJmoEQLSE= golang.org/x/time v0.12.0/go.mod h1:CDIdPxbZBQxdj6cxyCIdrNogrJKMJ7pr37NYpMcMDSg= -golang.org/x/tools v0.40.0 h1:yLkxfA+Qnul4cs9QA3KnlFu0lVmd8JJfoq+E41uSutA= -golang.org/x/tools v0.40.0/go.mod h1:Ik/tzLRlbscWpqqMRjyWYDisX8bG13FrdXp3o4Sr9lc= +golang.org/x/tools v0.42.0 h1:uNgphsn75Tdz5Ji2q36v/nsFSfR/9BRFvqhGBaJGd5k= +golang.org/x/tools v0.42.0/go.mod h1:Ma6lCIwGZvHK6XtgbswSoWroEkhugApmsXyrUmBhfr0= golang.org/x/tools/go/expect v0.1.1-deprecated h1:jpBZDwmgPhXsKZC6WhL20P4b/wmnpsEAGHaNy0n/rJM= golang.org/x/tools/go/expect v0.1.1-deprecated/go.mod h1:eihoPOH+FgIqa3FpoTwguz/bVUSGBlGQU67vpBeOrBY= golang.org/x/tools/go/packages/packagestest v0.1.1-deprecated h1:1h2MnaIAIXISqTFKdENegdpAgUXz6NrPEsbIeWaBRvM= golang.org/x/tools/go/packages/packagestest v0.1.1-deprecated/go.mod h1:RVAQXBGNv1ib0J382/DPCRS/BPnsGebyM1Gj5VSDpG8= gomodules.xyz/jsonpatch/v2 v2.4.0 h1:Ci3iUJyx9UeRx7CeFN8ARgGbkESwJK+KB9lLcWxY/Zw= gomodules.xyz/jsonpatch/v2 v2.4.0/go.mod h1:AH3dM2RI6uoBZxn3LVrfvJ3E0/9dG4cSrbuBJT4moAY= -google.golang.org/genproto/googleapis/api v0.0.0-20250303144028-a0af3efb3deb h1:p31xT4yrYrSM/G4Sn2+TNUkVhFCbG9y8itM2S6Th950= -google.golang.org/genproto/googleapis/api v0.0.0-20250303144028-a0af3efb3deb/go.mod h1:jbe3Bkdp+Dh2IrslsFCklNhweNTBgSYanP1UXhJDhKg= -google.golang.org/genproto/googleapis/rpc v0.0.0-20250528174236-200df99c418a h1:v2PbRU4K3llS09c7zodFpNePeamkAwG3mPrAery9VeE= -google.golang.org/genproto/googleapis/rpc v0.0.0-20250528174236-200df99c418a/go.mod h1:qQ0YXyHHx3XkvlzUtpXDkS29lDSafHMZBAZDc03LQ3A= -google.golang.org/grpc v1.72.2 h1:TdbGzwb82ty4OusHWepvFWGLgIbNo1/SUynEN0ssqv8= -google.golang.org/grpc v1.72.2/go.mod h1:wH5Aktxcg25y1I3w7H69nHfXdOG3UiadoBtjh3izSDM= -google.golang.org/protobuf v1.36.8 h1:xHScyCOEuuwZEc6UtSOvPbAT4zRh0xcNRYekJwfqyMc= -google.golang.org/protobuf v1.36.8/go.mod h1:fuxRtAxBytpl4zzqUh6/eyUujkJdNiuEkXntxiD/uRU= +gonum.org/v1/gonum v0.17.0 h1:VbpOemQlsSMrYmn7T2OUvQ4dqxQXU+ouZFQsZOx50z4= +gonum.org/v1/gonum v0.17.0/go.mod h1:El3tOrEuMpv2UdMrbNlKEh9vd86bmQ6vqIcDwxEOc1E= +google.golang.org/genproto/googleapis/api v0.0.0-20260401024825-9d38bb4040a9 h1:VPWxll4HlMw1Vs/qXtN7BvhZqsS9cdAittCNvVENElA= +google.golang.org/genproto/googleapis/api v0.0.0-20260401024825-9d38bb4040a9/go.mod h1:7QBABkRtR8z+TEnmXTqIqwJLlzrZKVfAUm7tY3yGv0M= +google.golang.org/genproto/googleapis/rpc v0.0.0-20260401024825-9d38bb4040a9 h1:m8qni9SQFH0tJc1X0vmnpw/0t+AImlSvp30sEupozUg= +google.golang.org/genproto/googleapis/rpc v0.0.0-20260401024825-9d38bb4040a9/go.mod h1:4Hqkh8ycfw05ld/3BWL7rJOSfebL2Q+DVDeRgYgxUU8= +google.golang.org/grpc v1.80.0 h1:Xr6m2WmWZLETvUNvIUmeD5OAagMw3FiKmMlTdViWsHM= +google.golang.org/grpc v1.80.0/go.mod h1:ho/dLnxwi3EDJA4Zghp7k2Ec1+c2jqup0bFkw07bwF4= +google.golang.org/protobuf v1.36.11 h1:fV6ZwhNocDyBLK0dj+fg8ektcVegBBuEolpbTQyBNVE= +google.golang.org/protobuf v1.36.11/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco= gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0= gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk= gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q= diff --git a/pkg/controller/options.go b/pkg/controller/options.go index 3c10c6d..047f9cf 100644 --- a/pkg/controller/options.go +++ b/pkg/controller/options.go @@ -37,12 +37,34 @@ type Options struct { // MaxConcurrentReconciles for each controller. MaxConcurrentReconciles int + + // UsePriorityQueue controls whether controller-runtime should use its + // priority queue implementation when queue wait telemetry is enabled. + // Defaults to true. + UsePriorityQueue *bool + + // QueueWaitRecorder records how long requests spend waiting in the + // controller queue before a worker starts processing them. + QueueWaitRecorder QueueWaitRecorder } // ForControllerRuntime extracts options for controller-runtime. func (o Options) ForControllerRuntime() controller.Options { - return controller.TypedOptions[reconcile.Request]{ + controllerOptions := controller.TypedOptions[reconcile.Request]{ MaxConcurrentReconciles: o.MaxConcurrentReconciles, RateLimiter: workqueue.NewTypedItemExponentialFailureRateLimiter[reconcile.Request](1*time.Second, 60*time.Second), } + + if o.QueueWaitRecorder != nil { + usePriorityQueue := true + if o.UsePriorityQueue != nil { + usePriorityQueue = *o.UsePriorityQueue + } + + controllerOptions.NewQueue = func(controllerName string, rateLimiter workqueue.TypedRateLimiter[reconcile.Request]) workqueue.TypedRateLimitingInterface[reconcile.Request] { + return newQueueWithWaitMetrics(controllerName, rateLimiter, usePriorityQueue, o.QueueWaitRecorder) + } + } + + return controllerOptions } diff --git a/pkg/controller/queue_wait.go b/pkg/controller/queue_wait.go new file mode 100644 index 0000000..2f033f2 --- /dev/null +++ b/pkg/controller/queue_wait.go @@ -0,0 +1,193 @@ +package controller + +import ( + "context" + "sync" + "time" + + "github.com/go-logr/logr" + "k8s.io/client-go/util/workqueue" + sigspq "sigs.k8s.io/controller-runtime/pkg/controller/priorityqueue" + "sigs.k8s.io/controller-runtime/pkg/reconcile" +) + +// QueueWaitRecorder records queue telemetry for a controller queue. +type QueueWaitRecorder interface { + RecordQueueWaitDuration(ctx context.Context, d time.Duration) + AddQueueDepth(ctx context.Context, delta int64) + RecordQueueOldestItemAge(ctx context.Context, d time.Duration) + RecordQueueWorkDuration(ctx context.Context, d time.Duration) +} + +func newQueueWithWaitMetrics(controllerName string, rateLimiter workqueue.TypedRateLimiter[reconcile.Request], usePriorityQueue bool, recorder QueueWaitRecorder) workqueue.TypedRateLimitingInterface[reconcile.Request] { + if recorder == nil { + return newControllerQueue(controllerName, rateLimiter, usePriorityQueue) + } + + return &queueWaitInstrumented[reconcile.Request]{ + queue: newControllerQueue(controllerName, rateLimiter, usePriorityQueue), + recorder: recorder, + enqueuedAt: make(map[reconcile.Request]time.Time), + startedAt: make(map[reconcile.Request]time.Time), + } +} + +func newControllerQueue(controllerName string, rateLimiter workqueue.TypedRateLimiter[reconcile.Request], usePriorityQueue bool) workqueue.TypedRateLimitingInterface[reconcile.Request] { + if usePriorityQueue { + return sigspq.New(controllerName, func(o *sigspq.Opts[reconcile.Request]) { + o.Log = logr.Discard() + o.RateLimiter = rateLimiter + }) + } + + return workqueue.NewTypedRateLimitingQueueWithConfig(rateLimiter, workqueue.TypedRateLimitingQueueConfig[reconcile.Request]{ + Name: controllerName, + }) +} + +type queueWaitInstrumented[T comparable] struct { + queue workqueue.TypedRateLimitingInterface[T] + recorder QueueWaitRecorder + lock sync.Mutex + enqueuedAt map[T]time.Time + startedAt map[T]time.Time +} + +func (q *queueWaitInstrumented[T]) Add(item T) { + q.recordEnqueueAt(item, time.Now()) + q.queue.Add(item) +} + +func (q *queueWaitInstrumented[T]) AddAfter(item T, duration time.Duration) { + q.recordEnqueueAt(item, time.Now()) + q.queue.AddAfter(item, duration) +} + +func (q *queueWaitInstrumented[T]) AddRateLimited(item T) { + q.recordEnqueueAt(item, time.Now()) + q.queue.AddRateLimited(item) +} + +func (q *queueWaitInstrumented[T]) Get() (item T, shutdown bool) { + item, shutdown = q.queue.Get() + if !shutdown { + q.recordDequeueAt(item, time.Now()) + } + return item, shutdown +} + +func (q *queueWaitInstrumented[T]) Done(item T) { + q.recordDoneAt(item, time.Now()) + q.queue.Done(item) +} + +func (q *queueWaitInstrumented[T]) ShutDown() { + if pending := q.resetState(); pending > 0 { + q.recorder.AddQueueDepth(context.Background(), -pending) + } + q.queue.ShutDown() +} + +func (q *queueWaitInstrumented[T]) ShutDownWithDrain() { + q.queue.ShutDownWithDrain() +} + +func (q *queueWaitInstrumented[T]) ShuttingDown() bool { + return q.queue.ShuttingDown() +} + +func (q *queueWaitInstrumented[T]) Len() int { + return q.queue.Len() +} + +func (q *queueWaitInstrumented[T]) Forget(item T) { + q.queue.Forget(item) +} + +func (q *queueWaitInstrumented[T]) NumRequeues(item T) int { + return q.queue.NumRequeues(item) +} + +func (q *queueWaitInstrumented[T]) recordEnqueueAt(item T, started time.Time) (bool, time.Duration) { + q.lock.Lock() + + if _, exists := q.enqueuedAt[item]; exists { + oldestAge := q.oldestItemAgeLocked(started) + q.lock.Unlock() + return false, oldestAge + } + + q.enqueuedAt[item] = started + oldestAge := q.oldestItemAgeLocked(started) + q.lock.Unlock() + + q.recorder.AddQueueDepth(context.Background(), 1) + q.recorder.RecordQueueOldestItemAge(context.Background(), oldestAge) + + return true, oldestAge +} + +func (q *queueWaitInstrumented[T]) recordDequeue(item T) (time.Duration, time.Duration, bool) { + return q.recordDequeueAt(item, time.Now()) +} + +func (q *queueWaitInstrumented[T]) recordDequeueAt(item T, finished time.Time) (time.Duration, time.Duration, bool) { + q.lock.Lock() + started, exists := q.enqueuedAt[item] + if exists { + delete(q.enqueuedAt, item) + q.startedAt[item] = finished + } + oldestAge := q.oldestItemAgeLocked(finished) + q.lock.Unlock() + + if !exists { + return 0, oldestAge, false + } + + q.recorder.AddQueueDepth(context.Background(), -1) + q.recorder.RecordQueueWaitDuration(context.Background(), finished.Sub(started)) + q.recorder.RecordQueueOldestItemAge(context.Background(), oldestAge) + + return finished.Sub(started), oldestAge, true +} + +func (q *queueWaitInstrumented[T]) recordDoneAt(item T, finished time.Time) (time.Duration, bool) { + q.lock.Lock() + started, exists := q.startedAt[item] + if exists { + delete(q.startedAt, item) + } + q.lock.Unlock() + + if !exists { + return 0, false + } + + workDuration := finished.Sub(started) + q.recorder.RecordQueueWorkDuration(context.Background(), workDuration) + + return workDuration, true +} + +func (q *queueWaitInstrumented[T]) resetState() int64 { + q.lock.Lock() + pending := int64(len(q.enqueuedAt)) + q.enqueuedAt = make(map[T]time.Time) + q.startedAt = make(map[T]time.Time) + q.lock.Unlock() + + return pending +} + +func (q *queueWaitInstrumented[T]) oldestItemAgeLocked(now time.Time) time.Duration { + var oldest time.Duration + for _, started := range q.enqueuedAt { + age := now.Sub(started) + if age > oldest { + oldest = age + } + } + + return oldest +} diff --git a/pkg/controller/queue_wait_test.go b/pkg/controller/queue_wait_test.go new file mode 100644 index 0000000..55befa3 --- /dev/null +++ b/pkg/controller/queue_wait_test.go @@ -0,0 +1,171 @@ +package controller + +import ( + "context" + "sync" + "testing" + "time" +) + +type fakeQueueWaitRecorder struct { + lock sync.Mutex + waitDurations []time.Duration + depthDeltas []int64 + oldestAges []time.Duration + workDurations []time.Duration +} + +func (r *fakeQueueWaitRecorder) RecordQueueWaitDuration(_ context.Context, d time.Duration) { + r.lock.Lock() + defer r.lock.Unlock() + + r.waitDurations = append(r.waitDurations, d) +} + +func (r *fakeQueueWaitRecorder) AddQueueDepth(_ context.Context, delta int64) { + r.lock.Lock() + defer r.lock.Unlock() + + r.depthDeltas = append(r.depthDeltas, delta) +} + +func (r *fakeQueueWaitRecorder) RecordQueueOldestItemAge(_ context.Context, d time.Duration) { + r.lock.Lock() + defer r.lock.Unlock() + + r.oldestAges = append(r.oldestAges, d) +} + +func (r *fakeQueueWaitRecorder) RecordQueueWorkDuration(_ context.Context, d time.Duration) { + r.lock.Lock() + defer r.lock.Unlock() + + r.workDurations = append(r.workDurations, d) +} + +func TestQueueWaitInstrumentedRecordsQueueState(t *testing.T) { + recorder := &fakeQueueWaitRecorder{} + queue := &queueWaitInstrumented[string]{ + recorder: recorder, + enqueuedAt: map[string]time.Time{}, + startedAt: map[string]time.Time{}, + } + + started := time.Unix(100, 0) + inserted, oldestAge := queue.recordEnqueueAt("foo", started) + if !inserted { + t.Fatal("expected first enqueue to be recorded") + } + if oldestAge != 0 { + t.Fatalf("expected oldest age after first enqueue = 0, got %v", oldestAge) + } + inserted, _ = queue.recordEnqueueAt("foo", started.Add(10*time.Second)) + if inserted { + t.Fatal("expected duplicate enqueue to be ignored") + } + waitDuration, oldestAge, removed := queue.recordDequeueAt("foo", started.Add(25*time.Second)) + if !removed { + t.Fatal("expected dequeue to remove pending item") + } + if waitDuration != 25*time.Second { + t.Fatalf("recorded queue wait = %v, want %v", waitDuration, 25*time.Second) + } + if oldestAge != 0 { + t.Fatalf("expected oldest age after dequeue = 0, got %v", oldestAge) + } + workDuration, ok := queue.recordDoneAt("foo", started.Add(40*time.Second)) + if !ok { + t.Fatal("expected processing duration to be recorded") + } + if workDuration != 15*time.Second { + t.Fatalf("recorded work duration = %v, want %v", workDuration, 15*time.Second) + } + + if len(recorder.waitDurations) != 1 { + t.Fatalf("expected 1 queue wait record, got %d", len(recorder.waitDurations)) + } + if got, want := recorder.waitDurations[0], 25*time.Second; got != want { + t.Fatalf("recorded queue wait = %v, want %v", got, want) + } + if got, want := recorder.depthDeltas, []int64{1, -1}; len(got) != len(want) { + t.Fatalf("expected %d depth updates, got %d", len(want), len(got)) + } else { + for i := range want { + if got[i] != want[i] { + t.Fatalf("depth delta[%d] = %d, want %d", i, got[i], want[i]) + } + } + } + if len(recorder.oldestAges) != 2 { + t.Fatalf("expected 2 oldest-age samples, got %d", len(recorder.oldestAges)) + } + if recorder.oldestAges[0] != 0 || recorder.oldestAges[1] != 0 { + t.Fatalf("expected oldest ages to remain at zero, got %v", recorder.oldestAges) + } + if len(recorder.workDurations) != 1 { + t.Fatalf("expected 1 work duration record, got %d", len(recorder.workDurations)) + } + if got, want := recorder.workDurations[0], 15*time.Second; got != want { + t.Fatalf("recorded work duration = %v, want %v", got, want) + } +} + +func TestQueueWaitInstrumentedIgnoresMissingEnqueue(t *testing.T) { + recorder := &fakeQueueWaitRecorder{} + queue := &queueWaitInstrumented[string]{ + recorder: recorder, + enqueuedAt: map[string]time.Time{}, + startedAt: map[string]time.Time{}, + } + + _, _, removed := queue.recordDequeueAt("foo", time.Unix(100, 0)) + if removed { + t.Fatal("expected missing enqueue to be ignored") + } + if workDuration, ok := queue.recordDoneAt("foo", time.Unix(125, 0)); ok || workDuration != 0 { + t.Fatalf("expected missing processing entry to be ignored, got duration=%v ok=%v", workDuration, ok) + } + + if len(recorder.waitDurations) != 0 { + t.Fatalf("expected no queue wait records, got %d", len(recorder.waitDurations)) + } + if len(recorder.depthDeltas) != 0 { + t.Fatalf("expected no depth updates, got %d", len(recorder.depthDeltas)) + } + if len(recorder.oldestAges) != 0 { + t.Fatalf("expected no oldest-age samples, got %d", len(recorder.oldestAges)) + } + if len(recorder.workDurations) != 0 { + t.Fatalf("expected no work duration records, got %d", len(recorder.workDurations)) + } +} + +func TestOptionsForControllerRuntimeQueueWaitRecorder(t *testing.T) { + tests := map[string]struct { + opts Options + wantNewQueue bool + }{ + "recorder disabled": { + opts: Options{}, + wantNewQueue: false, + }, + "recorder enabled": { + opts: Options{ + QueueWaitRecorder: &fakeQueueWaitRecorder{}, + }, + wantNewQueue: true, + }, + } + + for name, tc := range tests { + t.Run(name, func(t *testing.T) { + runtimeOpts := tc.opts.ForControllerRuntime() + if tc.wantNewQueue && runtimeOpts.NewQueue == nil { + t.Fatal("expected custom NewQueue to be configured") + } + if !tc.wantNewQueue && runtimeOpts.NewQueue != nil { + t.Fatal("did not expect a custom NewQueue to be configured") + } + }) + } +} diff --git a/pkg/reconciler/reconciler.go b/pkg/reconciler/reconciler.go index d45199c..87d90b3 100644 --- a/pkg/reconciler/reconciler.go +++ b/pkg/reconciler/reconciler.go @@ -90,6 +90,37 @@ func ControllerName(kind string) string { return "managed/" + strings.ToLower(kind) } +// MetricsRecorder records high-level reconcile metrics. +type MetricsRecorder interface { + RecordGetDuration(ctx context.Context, d time.Duration) + IncGetFailure(ctx context.Context) + + RecordConnectDuration(ctx context.Context, d time.Duration) + IncConnectFailure(ctx context.Context) + RecordObserveDuration(ctx context.Context, d time.Duration) + IncObserveFailure(ctx context.Context) + RecordCreateDuration(ctx context.Context, d time.Duration) + IncCreateFailure(ctx context.Context) + RecordUpdateDuration(ctx context.Context, d time.Duration) + IncUpdateFailure(ctx context.Context) + RecordDeleteDuration(ctx context.Context, d time.Duration) + IncDeleteFailure(ctx context.Context) + RecordFinalizerAddDuration(ctx context.Context, d time.Duration) + IncFinalizerAddFailure(ctx context.Context) + RecordFinalizerRemoveDuration(ctx context.Context, d time.Duration) + IncFinalizerRemoveFailure(ctx context.Context) + RecordStatusUpdateDuration(ctx context.Context, d time.Duration) + IncStatusUpdateFailure(ctx context.Context) + + RecordReconcileDuration(ctx context.Context, d time.Duration) + IncReconcileSuccess(ctx context.Context) + IncReconcileFailure(ctx context.Context) + IncReconcileRequeueAfter(ctx context.Context) + IncReconcileRequeueImmediate(ctx context.Context) + IncReconcileErrorRequeue(ctx context.Context) + AddReconcileInFlight(delta int64) +} + // A CriticalAnnotationUpdater is used when it is critical that annotations must // be updated before returning from the Reconcile loop. type CriticalAnnotationUpdater interface { @@ -340,6 +371,7 @@ type Reconciler struct { log logging.Logger record event.Recorder throttledRecorder event.Recorder + metrics MetricsRecorder } type mrManaged struct { @@ -373,6 +405,96 @@ func defaultPollIntervalHook(_ resource.Managed, pollInterval time.Duration) tim return pollInterval } +func (r *Reconciler) recordGet(ctx context.Context, started time.Time, err error) { + if r.metrics == nil { + return + } + r.metrics.RecordGetDuration(ctx, time.Since(started)) + if err != nil { + r.metrics.IncGetFailure(ctx) + } +} + +func (r *Reconciler) recordConnect(ctx context.Context, started time.Time, err error) { + if r.metrics == nil { + return + } + r.metrics.RecordConnectDuration(ctx, time.Since(started)) + if err != nil { + r.metrics.IncConnectFailure(ctx) + } +} + +func (r *Reconciler) recordObserve(ctx context.Context, started time.Time, err error) { + if r.metrics == nil { + return + } + r.metrics.RecordObserveDuration(ctx, time.Since(started)) + if err != nil { + r.metrics.IncObserveFailure(ctx) + } +} + +func (r *Reconciler) recordCreate(ctx context.Context, started time.Time, err error) { + if r.metrics == nil { + return + } + r.metrics.RecordCreateDuration(ctx, time.Since(started)) + if err != nil { + r.metrics.IncCreateFailure(ctx) + } +} + +func (r *Reconciler) recordUpdate(ctx context.Context, started time.Time, err error) { + if r.metrics == nil { + return + } + r.metrics.RecordUpdateDuration(ctx, time.Since(started)) + if err != nil { + r.metrics.IncUpdateFailure(ctx) + } +} + +func (r *Reconciler) recordDelete(ctx context.Context, started time.Time, err error) { + if r.metrics == nil { + return + } + r.metrics.RecordDeleteDuration(ctx, time.Since(started)) + if err != nil { + r.metrics.IncDeleteFailure(ctx) + } +} + +func (r *Reconciler) recordFinalizerAdd(ctx context.Context, started time.Time, err error) { + if r.metrics == nil { + return + } + r.metrics.RecordFinalizerAddDuration(ctx, time.Since(started)) + if err != nil { + r.metrics.IncFinalizerAddFailure(ctx) + } +} + +func (r *Reconciler) recordFinalizerRemove(ctx context.Context, started time.Time, err error) { + if r.metrics == nil { + return + } + r.metrics.RecordFinalizerRemoveDuration(ctx, time.Since(started)) + if err != nil { + r.metrics.IncFinalizerRemoveFailure(ctx) + } +} + +func (r *Reconciler) recordStatusUpdate(ctx context.Context, started time.Time, err error) { + if r.metrics == nil { + return + } + r.metrics.RecordStatusUpdateDuration(ctx, time.Since(started)) + if err != nil { + r.metrics.IncStatusUpdateFailure(ctx) + } +} + // A ReconcilerOption configures a Reconciler. type ReconcilerOption func(*Reconciler) @@ -470,6 +592,13 @@ func WithLogger(l logging.Logger) ReconcilerOption { } } +// WithMetrics specifies how the Reconciler should record telemetry. +func WithMetrics(m MetricsRecorder) ReconcilerOption { + return func(r *Reconciler) { + r.metrics = m + } +} + // WithRecorder specifies how the Reconciler should record events. func WithRecorder(er event.Recorder) ReconcilerOption { return func(r *Reconciler) { @@ -526,7 +655,7 @@ func NewReconciler(m manager.Manager, of resource.ManagedKind, o ...ReconcilerOp } // Reconcile a managed resource with an external resource. -func (r *Reconciler) Reconcile(ctx context.Context, req reconcile.Request) (reconcile.Result, error) { //nolint:gocyclo // See note below. +func (r *Reconciler) Reconcile(ctx context.Context, req reconcile.Request) (result reconcile.Result, err error) { //nolint:gocyclo // See note below. log := r.log.WithValues("request", req) log.Debug("Reconciling") @@ -536,8 +665,39 @@ func (r *Reconciler) Reconcile(ctx context.Context, req reconcile.Request) (reco externalCtx, externalCancel := context.WithTimeout(ctx, r.timeout) defer externalCancel() + statusUpdate := func(updateFn func() error) error { + started := time.Now() + updateErr := updateFn() + r.recordStatusUpdate(ctx, started, updateErr) + return updateErr + } + + started := time.Now() + if r.metrics != nil { + r.metrics.AddReconcileInFlight(1) + defer r.metrics.AddReconcileInFlight(-1) + defer func() { + r.metrics.RecordReconcileDuration(ctx, time.Since(started)) + if err != nil { + r.metrics.IncReconcileFailure(ctx) + r.metrics.IncReconcileErrorRequeue(ctx) + return + } + + r.metrics.IncReconcileSuccess(ctx) + switch { + case result.RequeueAfter > 0: + r.metrics.IncReconcileRequeueAfter(ctx) + case result.Requeue: + r.metrics.IncReconcileRequeueImmediate(ctx) + } + }() + } + managed := r.newManaged() - if err := r.client.Get(ctx, req.NamespacedName, managed); err != nil { + err = r.client.Get(ctx, req.NamespacedName, managed) + r.recordGet(ctx, started, err) + if err != nil { // There's no need to requeue if we no longer exist. Otherwise we'll be // requeued implicitly because we return an error. log.Error(err, "Cannot get managed resource") @@ -560,7 +720,10 @@ func (r *Reconciler) Reconcile(ctx context.Context, req reconcile.Request) (reco managed.SetConditions(prv1.ReconcilePaused()) // if the pause annotation is removed, we will have a chance to reconcile again and resume // and if status update fails, we will reconcile again to retry to update the status - return reconcile.Result{}, errors.Wrap(r.client.Status().Update(ctx, managed), errUpdateManagedStatus) + if err := statusUpdate(func() error { return r.client.Status().Update(ctx, managed) }); err != nil { + return reconcile.Result{}, errors.Wrap(err, errUpdateManagedStatus) + } + return reconcile.Result{}, nil } // If managed resource has a deletion timestamp and and a deletion policy of @@ -569,7 +732,10 @@ func (r *Reconciler) Reconcile(ctx context.Context, req reconcile.Request) (reco if meta.WasDeleted(managed) && !meta.ShouldDelete(managed) { log = log.WithValues("deletion-timestamp", managed.GetDeletionTimestamp()) - if err := r.managed.RemoveFinalizer(ctx, managed); err != nil { + started = time.Now() + err = r.managed.RemoveFinalizer(ctx, managed) + r.recordFinalizerRemove(ctx, started, err) + if err != nil { // If this is the first time we encounter this issue we'll be // requeued implicitly when we update our status with the new error // condition. If not, we requeue explicitly, which will trigger @@ -579,7 +745,10 @@ func (r *Reconciler) Reconcile(ctx context.Context, req reconcile.Request) (reco return reconcile.Result{Requeue: true}, nil } managed.SetConditions(prv1.Deleting(), prv1.ReconcileError(err)) - return reconcile.Result{Requeue: true}, errors.Wrap(r.client.Status().Update(ctx, managed), errUpdateManagedStatus) + if err := statusUpdate(func() error { return r.client.Status().Update(ctx, managed) }); err != nil { + return reconcile.Result{Requeue: true}, errors.Wrap(err, errUpdateManagedStatus) + } + return reconcile.Result{Requeue: true}, nil } // We've successfully unpublished our managed resource's connection @@ -598,10 +767,15 @@ func (r *Reconciler) Reconcile(ctx context.Context, req reconcile.Request) (reco log.Warn(errCreateIncomplete) throttledRecorder.Event(managed, event.Warning(reasonCannotInitialize, actionProcessEvent, errors.New(errCreateIncomplete))) managed.SetConditions(prv1.Creating(), prv1.ReconcileError(errors.New(errCreateIncomplete))) - return reconcile.Result{Requeue: false}, errors.Wrap(r.client.Status().Update(ctx, managed), errUpdateManagedStatus) + if err := statusUpdate(func() error { return r.client.Status().Update(ctx, managed) }); err != nil { + return reconcile.Result{Requeue: false}, errors.Wrap(err, errUpdateManagedStatus) + } + return reconcile.Result{Requeue: false}, nil } + started = time.Now() external, err := r.external.Connect(externalCtx, managed) + r.recordConnect(ctx, started, err) if err != nil { // We'll usually hit this case if our Provider or its secret are missing // or invalid. If this is first time we encounter this issue we'll be @@ -614,7 +788,10 @@ func (r *Reconciler) Reconcile(ctx context.Context, req reconcile.Request) (reco } throttledRecorder.Event(managed, event.Warning(reasonCannotConnect, actionProcessEvent, err)) managed.SetConditions(prv1.ReconcileError(errors.Wrap(err, errReconcileConnect))) - return reconcile.Result{Requeue: true}, errors.Wrap(r.client.Status().Update(ctx, managed), errUpdateManagedStatus) + if err := statusUpdate(func() error { return r.client.Status().Update(ctx, managed) }); err != nil { + return reconcile.Result{Requeue: true}, errors.Wrap(err, errUpdateManagedStatus) + } + return reconcile.Result{Requeue: true}, nil } defer func() { if err := r.external.Disconnect(ctx); err != nil { @@ -623,7 +800,9 @@ func (r *Reconciler) Reconcile(ctx context.Context, req reconcile.Request) (reco } }() + started = time.Now() observation, err := external.Observe(externalCtx, managed) + r.recordObserve(ctx, started, err) if err != nil { // We'll usually hit this case if our Provider credentials are invalid // or insufficient for observing the external resource type we're @@ -637,7 +816,10 @@ func (r *Reconciler) Reconcile(ctx context.Context, req reconcile.Request) (reco } throttledRecorder.Event(managed, event.Warning(reasonCannotObserve, actionObserveExternalResource, err)) managed.SetConditions(prv1.ReconcileError(errors.Wrap(err, errReconcileObserve))) - return reconcile.Result{Requeue: true}, errors.Wrap(r.client.Status().Update(ctx, managed), errUpdateManagedStatus) + if err := statusUpdate(func() error { return r.client.Status().Update(ctx, managed) }); err != nil { + return reconcile.Result{Requeue: true}, errors.Wrap(err, errUpdateManagedStatus) + } + return reconcile.Result{Requeue: true}, nil } // In the observe-only mode, !observation.ResourceExists will be an error @@ -645,7 +827,10 @@ func (r *Reconciler) Reconcile(ctx context.Context, req reconcile.Request) (reco if !observation.ResourceExists && meta.ShouldOnlyObserve(managed) { throttledRecorder.Event(managed, event.Warning(reasonCannotObserve, actionObserveExternalResource, errors.New(errExternalResourceNotExist))) managed.SetConditions(prv1.ReconcileError(errors.Wrap(errors.New(errExternalResourceNotExist), errReconcileObserve))) - return reconcile.Result{Requeue: true}, errors.Wrap(r.client.Status().Update(ctx, managed), errUpdateManagedStatus) + if err := statusUpdate(func() error { return r.client.Status().Update(ctx, managed) }); err != nil { + return reconcile.Result{Requeue: true}, errors.Wrap(err, errUpdateManagedStatus) + } + return reconcile.Result{Requeue: true}, nil } // If this resource has a non-zero creation grace period we want to wait @@ -665,7 +850,10 @@ func (r *Reconciler) Reconcile(ctx context.Context, req reconcile.Request) (reco // We'll only reach this point if deletion policy is not orphan, so we // are safe to call external deletion if external resource exists. if observation.ResourceExists && meta.ShouldDelete(managed) { - if err := external.Delete(externalCtx, managed); err != nil { + started = time.Now() + err = external.Delete(externalCtx, managed) + r.recordDelete(ctx, started, err) + if err != nil { // We'll hit this condition if we can't delete our external // resource, for example if our provider credentials don't have // access to delete it. If this is the first time we encounter @@ -675,7 +863,10 @@ func (r *Reconciler) Reconcile(ctx context.Context, req reconcile.Request) (reco log.Error(err, "Cannot delete external resource") throttledRecorder.Event(managed, event.Warning(reasonCannotDelete, actionDeleteExternalResource, err)) managed.SetConditions(prv1.Deleting(), prv1.ReconcileError(errors.Wrap(err, errReconcileDelete))) - return reconcile.Result{Requeue: true}, errors.Wrap(r.client.Status().Update(ctx, managed), errUpdateManagedStatus) + if err := statusUpdate(func() error { return r.client.Status().Update(ctx, managed) }); err != nil { + return reconcile.Result{Requeue: true}, errors.Wrap(err, errUpdateManagedStatus) + } + return reconcile.Result{Requeue: true}, nil } // We've successfully requested deletion of our external resource. @@ -688,10 +879,16 @@ func (r *Reconciler) Reconcile(ctx context.Context, req reconcile.Request) (reco log.Info("Successfully requested deletion of external resource") record.Event(managed, event.Normal(reasonDeleted, actionDeleteEvent, "Successfully requested deletion of external resource")) managed.SetConditions(prv1.Deleting(), prv1.ReconcileSuccess()) - return reconcile.Result{Requeue: true}, errors.Wrap(r.client.Status().Update(ctx, managed), errUpdateManagedStatus) + if err := statusUpdate(func() error { return r.client.Status().Update(ctx, managed) }); err != nil { + return reconcile.Result{Requeue: true}, errors.Wrap(err, errUpdateManagedStatus) + } + return reconcile.Result{Requeue: true}, nil } - if err := r.managed.RemoveFinalizer(ctx, managed); err != nil { + started = time.Now() + err = r.managed.RemoveFinalizer(ctx, managed) + r.recordFinalizerRemove(ctx, started, err) + if err != nil { // If this is the first time we encounter this issue we'll be // requeued implicitly when we update our status with the new error // condition. If not, we requeue explicitly, which will trigger @@ -701,7 +898,10 @@ func (r *Reconciler) Reconcile(ctx context.Context, req reconcile.Request) (reco return reconcile.Result{Requeue: true}, nil } managed.SetConditions(prv1.Deleting(), prv1.ReconcileError(err)) - return reconcile.Result{Requeue: true}, errors.Wrap(r.client.Status().Update(ctx, managed), errUpdateManagedStatus) + if err := statusUpdate(func() error { return r.client.Status().Update(ctx, managed) }); err != nil { + return reconcile.Result{Requeue: true}, errors.Wrap(err, errUpdateManagedStatus) + } + return reconcile.Result{Requeue: true}, nil } // We've successfully deleted our external resource (if necessary) and @@ -712,7 +912,10 @@ func (r *Reconciler) Reconcile(ctx context.Context, req reconcile.Request) (reco return reconcile.Result{Requeue: false}, nil } - if err := r.managed.AddFinalizer(ctx, managed); err != nil { + started = time.Now() + err = r.managed.AddFinalizer(ctx, managed) + r.recordFinalizerAdd(ctx, started, err) + if err != nil { // If this is the first time we encounter this issue we'll be requeued // implicitly when we update our status with the new error condition. If // not, we requeue explicitly, which will trigger backoff. @@ -721,7 +924,10 @@ func (r *Reconciler) Reconcile(ctx context.Context, req reconcile.Request) (reco return reconcile.Result{Requeue: true}, nil } managed.SetConditions(prv1.ReconcileError(err)) - return reconcile.Result{Requeue: true}, errors.Wrap(r.client.Status().Update(ctx, managed), errUpdateManagedStatus) + if err := statusUpdate(func() error { return r.client.Status().Update(ctx, managed) }); err != nil { + return reconcile.Result{Requeue: true}, errors.Wrap(err, errUpdateManagedStatus) + } + return reconcile.Result{Requeue: true}, nil } if !observation.ResourceExists && meta.ShouldCreate(managed) { @@ -740,10 +946,15 @@ func (r *Reconciler) Reconcile(ctx context.Context, req reconcile.Request) (reco } throttledRecorder.Event(managed, event.Warning(reasonCannotUpdateManaged, actionUpdateExternalResource, errors.Wrap(err, errUpdateManaged))) managed.SetConditions(prv1.Creating(), prv1.ReconcileError(errors.Wrap(err, errUpdateManaged))) - return reconcile.Result{Requeue: true}, errors.Wrap(r.client.Status().Update(ctx, managed), errUpdateManagedStatus) + if err := statusUpdate(func() error { return r.client.Status().Update(ctx, managed) }); err != nil { + return reconcile.Result{Requeue: true}, errors.Wrap(err, errUpdateManagedStatus) + } + return reconcile.Result{Requeue: true}, nil } + started = time.Now() err = external.Create(externalCtx, managed) + r.recordCreate(ctx, started, err) if err != nil { // We'll hit this condition if we can't create our external // resource, for example if our provider credentials don't have @@ -775,7 +986,10 @@ func (r *Reconciler) Reconcile(ctx context.Context, req reconcile.Request) (reco } managed.SetConditions(prv1.Creating(), prv1.ReconcileError(errors.Wrap(err, errReconcileCreate))) - return reconcile.Result{Requeue: true}, errors.Wrap(r.client.Status().Update(ctx, managed), errUpdateManagedStatus) + if err := statusUpdate(func() error { return r.client.Status().Update(ctx, managed) }); err != nil { + return reconcile.Result{Requeue: true}, errors.Wrap(err, errUpdateManagedStatus) + } + return reconcile.Result{Requeue: true}, nil } // In some cases our external-name may be set by Create above. @@ -797,7 +1011,10 @@ func (r *Reconciler) Reconcile(ctx context.Context, req reconcile.Request) (reco log.Error(err, errUpdateManagedAnnotations) throttledRecorder.Event(managed, event.Warning(reasonCannotUpdateManaged, actionUpdateExternalResource, errors.Wrap(err, errUpdateManagedAnnotations))) managed.SetConditions(prv1.Creating(), prv1.ReconcileError(errors.Wrap(err, errUpdateManagedAnnotations))) - return reconcile.Result{Requeue: true}, errors.Wrap(r.client.Status().Update(ctx, managed), errUpdateManagedStatus) + if err := statusUpdate(func() error { return r.client.Status().Update(ctx, managed) }); err != nil { + return reconcile.Result{Requeue: true}, errors.Wrap(err, errUpdateManagedStatus) + } + return reconcile.Result{Requeue: true}, nil } // We've successfully created our external resource. In many cases the @@ -807,7 +1024,10 @@ func (r *Reconciler) Reconcile(ctx context.Context, req reconcile.Request) (reco log.Info("Successfully requested creation of external resource") record.Event(managed, event.Normal(reasonCreated, actionCreateEvent, "Successfully requested creation of external resource")) managed.SetConditions(prv1.Creating(), prv1.ReconcileSuccess()) - return reconcile.Result{Requeue: true}, errors.Wrap(r.client.Status().Update(ctx, managed), errUpdateManagedStatus) + if err := statusUpdate(func() error { return r.client.Status().Update(ctx, managed) }); err != nil { + return reconcile.Result{Requeue: true}, errors.Wrap(err, errUpdateManagedStatus) + } + return reconcile.Result{Requeue: true}, nil } if observation.ResourceLateInitialized { @@ -822,7 +1042,10 @@ func (r *Reconciler) Reconcile(ctx context.Context, req reconcile.Request) (reco log.Error(err, "Cannot update managed resource") throttledRecorder.Event(managed, event.Warning(reasonCannotUpdateManaged, actionUpdateExternalResource, err)) managed.SetConditions(prv1.ReconcileError(errors.Wrap(err, errUpdateManaged))) - return reconcile.Result{Requeue: true}, errors.Wrap(r.client.Status().Update(ctx, managed), errUpdateManagedStatus) + if err := statusUpdate(func() error { return r.client.Status().Update(ctx, managed) }); err != nil { + return reconcile.Result{Requeue: true}, errors.Wrap(err, errUpdateManagedStatus) + } + return reconcile.Result{Requeue: true}, nil } } @@ -836,7 +1059,10 @@ func (r *Reconciler) Reconcile(ctx context.Context, req reconcile.Request) (reco reconcileAfter := r.pollIntervalHook(managed, r.pollInterval) log.Info("External resource is up to date", "requeue-after", time.Now().Add(reconcileAfter)) managed.SetConditions(prv1.ReconcileSuccess()) - return reconcile.Result{RequeueAfter: reconcileAfter}, errors.Wrap(r.client.Status().Update(ctx, managed), errUpdateManagedStatus) + if err := statusUpdate(func() error { return r.client.Status().Update(ctx, managed) }); err != nil { + return reconcile.Result{RequeueAfter: reconcileAfter}, errors.Wrap(err, errUpdateManagedStatus) + } + return reconcile.Result{RequeueAfter: reconcileAfter}, nil } if observation.Diff != "" { @@ -848,10 +1074,15 @@ func (r *Reconciler) Reconcile(ctx context.Context, req reconcile.Request) (reco reconcileAfter := r.pollIntervalHook(managed, r.pollInterval) log.Debug("Skipping update due to managementPolicies. Reconciliation succeeded", "requeue-after", time.Now().Add(reconcileAfter)) managed.SetConditions(prv1.ReconcileSuccess()) - return reconcile.Result{RequeueAfter: reconcileAfter}, errors.Wrap(r.client.Status().Update(ctx, managed), errUpdateManagedStatus) + if err := statusUpdate(func() error { return r.client.Status().Update(ctx, managed) }); err != nil { + return reconcile.Result{RequeueAfter: reconcileAfter}, errors.Wrap(err, errUpdateManagedStatus) + } + return reconcile.Result{RequeueAfter: reconcileAfter}, nil } + started = time.Now() err = external.Update(externalCtx, managed) + r.recordUpdate(ctx, started, err) if err != nil { // We'll hit this condition if we can't update our external resource, // for example if our provider credentials don't have access to update @@ -861,7 +1092,10 @@ func (r *Reconciler) Reconcile(ctx context.Context, req reconcile.Request) (reco log.Error(err, "Cannot update external resource") throttledRecorder.Event(managed, event.Warning(reasonCannotUpdate, actionUpdateExternalResource, err)) managed.SetConditions(prv1.ReconcileError(errors.Wrap(err, errReconcileUpdate))) - return reconcile.Result{Requeue: true}, errors.Wrap(r.client.Status().Update(ctx, managed), errUpdateManagedStatus) + if err := statusUpdate(func() error { return r.client.Status().Update(ctx, managed) }); err != nil { + return reconcile.Result{Requeue: true}, errors.Wrap(err, errUpdateManagedStatus) + } + return reconcile.Result{Requeue: true}, nil } // We've successfully updated our external resource. Per the below issue @@ -873,5 +1107,8 @@ func (r *Reconciler) Reconcile(ctx context.Context, req reconcile.Request) (reco log.Info("Successfully requested update of external resource", "requeue-after", time.Now().Add(reconcileAfter)) record.Event(managed, event.Normal(reasonUpdated, actionUpdateEvent, "Successfully requested update of external resource")) managed.SetConditions(prv1.ReconcileSuccess()) - return reconcile.Result{RequeueAfter: reconcileAfter}, errors.Wrap(r.client.Status().Update(ctx, managed), errUpdateManagedStatus) + if err := statusUpdate(func() error { return r.client.Status().Update(ctx, managed) }); err != nil { + return reconcile.Result{RequeueAfter: reconcileAfter}, errors.Wrap(err, errUpdateManagedStatus) + } + return reconcile.Result{RequeueAfter: reconcileAfter}, nil } diff --git a/pkg/telemetry/metrics.go b/pkg/telemetry/metrics.go new file mode 100644 index 0000000..e9af16c --- /dev/null +++ b/pkg/telemetry/metrics.go @@ -0,0 +1,455 @@ +package telemetry + +import ( + "context" + "sync/atomic" + "time" + + "github.com/krateoplatformops/provider-runtime/pkg/logging" + "go.opentelemetry.io/otel" + "go.opentelemetry.io/otel/attribute" + "go.opentelemetry.io/otel/exporters/otlp/otlpmetric/otlpmetrichttp" + "go.opentelemetry.io/otel/metric" + sdkmetric "go.opentelemetry.io/otel/sdk/metric" + "go.opentelemetry.io/otel/sdk/resource" +) + +const ( + defaultServiceName = "provider-runtime" + defaultExportInterval = 30 * time.Second +) + +// Config controls OpenTelemetry metrics export. +type Config struct { + Enabled bool + ServiceName string + ExportInterval time.Duration +} + +// Metrics exposes a small set of low-cardinality counters and histograms for +// provider-runtime consumers. +type Metrics struct { + startupSuccess metric.Int64Counter + startupFailure metric.Int64Counter + + getDuration metric.Float64Histogram + getFailure metric.Int64Counter + + reconcileDuration metric.Float64Histogram + queueDepth metric.Int64UpDownCounter + queueWaitDuration metric.Float64Histogram + queueOldestItemAge metric.Float64Histogram + queueWorkDuration metric.Float64Histogram + queueRequeueTotal metric.Int64Counter + reconcileSuccess metric.Int64Counter + reconcileFailure metric.Int64Counter + reconcileRequeueAfter metric.Int64Counter + reconcileRequeueImmediate metric.Int64Counter + reconcileErrorRequeue metric.Int64Counter + reconcileInFlight metric.Int64ObservableGauge + reconcileInFlightCount atomic.Int64 + + connectDuration metric.Float64Histogram + connectFailure metric.Int64Counter + + observeDuration metric.Float64Histogram + observeFailure metric.Int64Counter + + createDuration metric.Float64Histogram + createFailure metric.Int64Counter + + updateDuration metric.Float64Histogram + updateFailure metric.Int64Counter + + deleteDuration metric.Float64Histogram + deleteFailure metric.Int64Counter + + finalizerAddDuration metric.Float64Histogram + finalizerAddFailure metric.Int64Counter + finalizerRemoveDuration metric.Float64Histogram + finalizerRemoveFailure metric.Int64Counter + + statusUpdateDuration metric.Float64Histogram + statusUpdateFailure metric.Int64Counter +} + +// Setup creates and configures an OTLP metrics pipeline. +// +// When metrics are disabled, Setup returns a nil Metrics handle and a no-op +// shutdown function. +func Setup(ctx context.Context, log logging.Logger, cfg Config) (*Metrics, func(context.Context) error, error) { + if log == nil { + log = logging.NewNopLogger() + } + + if !cfg.Enabled { + return nil, func(context.Context) error { return nil }, nil + } + + serviceName := cfg.ServiceName + if serviceName == "" { + serviceName = defaultServiceName + } + + exportInterval := cfg.ExportInterval + if exportInterval <= 0 { + exportInterval = defaultExportInterval + } + + exporter, err := otlpmetrichttp.New(ctx) + if err != nil { + return nil, nil, err + } + + res, err := resource.Merge(resource.Default(), + resource.NewSchemaless(attribute.String("service.name", serviceName))) + if err != nil { + return nil, nil, err + } + + reader := sdkmetric.NewPeriodicReader(exporter, sdkmetric.WithInterval(exportInterval)) + provider := sdkmetric.NewMeterProvider( + sdkmetric.WithReader(reader), + sdkmetric.WithResource(res), + ) + + meter := provider.Meter("github.com/krateoplatformops/provider-runtime") + metrics, err := newMetrics(meter) + if err != nil { + _ = provider.Shutdown(ctx) + return nil, nil, err + } + + otel.SetMeterProvider(provider) + log.Info("OpenTelemetry metrics initialized") + + return metrics, provider.Shutdown, nil +} + +func newMetrics(meter metric.Meter) (*Metrics, error) { + var err error + m := &Metrics{} + + if m.startupSuccess, err = meter.Int64Counter("provider_runtime.startup.success"); err != nil { + return nil, err + } + if m.startupFailure, err = meter.Int64Counter("provider_runtime.startup.failure"); err != nil { + return nil, err + } + if m.getDuration, err = meter.Float64Histogram("provider_runtime.reconcile.get.duration_seconds"); err != nil { + return nil, err + } + if m.getFailure, err = meter.Int64Counter("provider_runtime.reconcile.get.failure"); err != nil { + return nil, err + } + if m.reconcileDuration, err = meter.Float64Histogram("provider_runtime.reconcile.duration_seconds"); err != nil { + return nil, err + } + if m.queueDepth, err = meter.Int64UpDownCounter("provider_runtime.reconcile.queue.depth"); err != nil { + return nil, err + } + if m.queueWaitDuration, err = meter.Float64Histogram("provider_runtime.reconcile.queue.wait.duration_seconds"); err != nil { + return nil, err + } + if m.queueOldestItemAge, err = meter.Float64Histogram("provider_runtime.reconcile.queue.oldest_item_age_seconds"); err != nil { + return nil, err + } + if m.queueWorkDuration, err = meter.Float64Histogram("provider_runtime.reconcile.queue.work.duration_seconds"); err != nil { + return nil, err + } + if m.queueRequeueTotal, err = meter.Int64Counter("provider_runtime.reconcile.queue.requeues"); err != nil { + return nil, err + } + if m.reconcileSuccess, err = meter.Int64Counter("provider_runtime.reconcile.success"); err != nil { + return nil, err + } + if m.reconcileFailure, err = meter.Int64Counter("provider_runtime.reconcile.failure"); err != nil { + return nil, err + } + if m.reconcileRequeueAfter, err = meter.Int64Counter("provider_runtime.reconcile.requeue.after"); err != nil { + return nil, err + } + if m.reconcileRequeueImmediate, err = meter.Int64Counter("provider_runtime.reconcile.requeue.immediate"); err != nil { + return nil, err + } + if m.reconcileErrorRequeue, err = meter.Int64Counter("provider_runtime.reconcile.requeue.error"); err != nil { + return nil, err + } + if m.reconcileInFlight, err = meter.Int64ObservableGauge("provider_runtime.reconcile.in_flight"); err != nil { + return nil, err + } + if m.connectDuration, err = meter.Float64Histogram("provider_runtime.external.connect.duration_seconds"); err != nil { + return nil, err + } + if m.connectFailure, err = meter.Int64Counter("provider_runtime.external.connect.failure"); err != nil { + return nil, err + } + if m.observeDuration, err = meter.Float64Histogram("provider_runtime.external.observe.duration_seconds"); err != nil { + return nil, err + } + if m.observeFailure, err = meter.Int64Counter("provider_runtime.external.observe.failure"); err != nil { + return nil, err + } + if m.createDuration, err = meter.Float64Histogram("provider_runtime.external.create.duration_seconds"); err != nil { + return nil, err + } + if m.createFailure, err = meter.Int64Counter("provider_runtime.external.create.failure"); err != nil { + return nil, err + } + if m.updateDuration, err = meter.Float64Histogram("provider_runtime.external.update.duration_seconds"); err != nil { + return nil, err + } + if m.updateFailure, err = meter.Int64Counter("provider_runtime.external.update.failure"); err != nil { + return nil, err + } + if m.deleteDuration, err = meter.Float64Histogram("provider_runtime.external.delete.duration_seconds"); err != nil { + return nil, err + } + if m.deleteFailure, err = meter.Int64Counter("provider_runtime.external.delete.failure"); err != nil { + return nil, err + } + if m.finalizerAddDuration, err = meter.Float64Histogram("provider_runtime.finalizer.add.duration_seconds"); err != nil { + return nil, err + } + if m.finalizerAddFailure, err = meter.Int64Counter("provider_runtime.finalizer.add.failure"); err != nil { + return nil, err + } + if m.finalizerRemoveDuration, err = meter.Float64Histogram("provider_runtime.finalizer.remove.duration_seconds"); err != nil { + return nil, err + } + if m.finalizerRemoveFailure, err = meter.Int64Counter("provider_runtime.finalizer.remove.failure"); err != nil { + return nil, err + } + if m.statusUpdateDuration, err = meter.Float64Histogram("provider_runtime.status.update.duration_seconds"); err != nil { + return nil, err + } + if m.statusUpdateFailure, err = meter.Int64Counter("provider_runtime.status.update.failure"); err != nil { + return nil, err + } + + _, err = meter.RegisterCallback(func(_ context.Context, observer metric.Observer) error { + observer.ObserveInt64(m.reconcileInFlight, m.reconcileInFlightCount.Load()) + return nil + }, m.reconcileInFlight) + if err != nil { + return nil, err + } + + return m, nil +} + +func (m *Metrics) IncStartupSuccess(ctx context.Context) { + if m == nil { + return + } + m.startupSuccess.Add(ctx, 1) +} + +func (m *Metrics) IncStartupFailure(ctx context.Context) { + if m == nil { + return + } + m.startupFailure.Add(ctx, 1) +} + +func (m *Metrics) RecordGetDuration(ctx context.Context, d time.Duration) { + if m == nil { + return + } + m.getDuration.Record(ctx, d.Seconds()) +} + +func (m *Metrics) IncGetFailure(ctx context.Context) { + if m == nil { + return + } + m.getFailure.Add(ctx, 1) +} + +func (m *Metrics) RecordReconcileDuration(ctx context.Context, d time.Duration) { + if m == nil { + return + } + m.reconcileDuration.Record(ctx, d.Seconds()) +} + +func (m *Metrics) RecordQueueWaitDuration(ctx context.Context, d time.Duration) { + if m == nil { + return + } + m.queueWaitDuration.Record(ctx, d.Seconds()) +} + +func (m *Metrics) AddQueueDepth(ctx context.Context, delta int64) { + if m == nil || delta == 0 { + return + } + m.queueDepth.Add(ctx, delta) +} + +func (m *Metrics) RecordQueueOldestItemAge(ctx context.Context, d time.Duration) { + if m == nil { + return + } + m.queueOldestItemAge.Record(ctx, d.Seconds()) +} + +func (m *Metrics) RecordQueueWorkDuration(ctx context.Context, d time.Duration) { + if m == nil { + return + } + m.queueWorkDuration.Record(ctx, d.Seconds()) +} + +func (m *Metrics) recordRequeue(ctx context.Context, reason string, counter metric.Int64Counter) { + if m == nil { + return + } + counter.Add(ctx, 1) + m.queueRequeueTotal.Add(ctx, 1, metric.WithAttributes(attribute.String("reason", reason))) +} + +func (m *Metrics) IncReconcileSuccess(ctx context.Context) { + if m == nil { + return + } + m.reconcileSuccess.Add(ctx, 1) +} + +func (m *Metrics) IncReconcileFailure(ctx context.Context) { + if m == nil { + return + } + m.reconcileFailure.Add(ctx, 1) +} + +func (m *Metrics) IncReconcileRequeueAfter(ctx context.Context) { + m.recordRequeue(ctx, "after", m.reconcileRequeueAfter) +} + +func (m *Metrics) IncReconcileRequeueImmediate(ctx context.Context) { + m.recordRequeue(ctx, "immediate", m.reconcileRequeueImmediate) +} + +func (m *Metrics) IncReconcileErrorRequeue(ctx context.Context) { + m.recordRequeue(ctx, "error", m.reconcileErrorRequeue) +} + +func (m *Metrics) AddReconcileInFlight(delta int64) { + if m == nil || delta == 0 { + return + } + m.reconcileInFlightCount.Add(delta) +} + +func (m *Metrics) RecordConnectDuration(ctx context.Context, d time.Duration) { + if m == nil { + return + } + m.connectDuration.Record(ctx, d.Seconds()) +} + +func (m *Metrics) IncConnectFailure(ctx context.Context) { + if m == nil { + return + } + m.connectFailure.Add(ctx, 1) +} + +func (m *Metrics) RecordObserveDuration(ctx context.Context, d time.Duration) { + if m == nil { + return + } + m.observeDuration.Record(ctx, d.Seconds()) +} + +func (m *Metrics) IncObserveFailure(ctx context.Context) { + if m == nil { + return + } + m.observeFailure.Add(ctx, 1) +} + +func (m *Metrics) RecordCreateDuration(ctx context.Context, d time.Duration) { + if m == nil { + return + } + m.createDuration.Record(ctx, d.Seconds()) +} + +func (m *Metrics) IncCreateFailure(ctx context.Context) { + if m == nil { + return + } + m.createFailure.Add(ctx, 1) +} + +func (m *Metrics) RecordUpdateDuration(ctx context.Context, d time.Duration) { + if m == nil { + return + } + m.updateDuration.Record(ctx, d.Seconds()) +} + +func (m *Metrics) IncUpdateFailure(ctx context.Context) { + if m == nil { + return + } + m.updateFailure.Add(ctx, 1) +} + +func (m *Metrics) RecordDeleteDuration(ctx context.Context, d time.Duration) { + if m == nil { + return + } + m.deleteDuration.Record(ctx, d.Seconds()) +} + +func (m *Metrics) IncDeleteFailure(ctx context.Context) { + if m == nil { + return + } + m.deleteFailure.Add(ctx, 1) +} + +func (m *Metrics) RecordFinalizerAddDuration(ctx context.Context, d time.Duration) { + if m == nil { + return + } + m.finalizerAddDuration.Record(ctx, d.Seconds()) +} + +func (m *Metrics) IncFinalizerAddFailure(ctx context.Context) { + if m == nil { + return + } + m.finalizerAddFailure.Add(ctx, 1) +} + +func (m *Metrics) RecordFinalizerRemoveDuration(ctx context.Context, d time.Duration) { + if m == nil { + return + } + m.finalizerRemoveDuration.Record(ctx, d.Seconds()) +} + +func (m *Metrics) IncFinalizerRemoveFailure(ctx context.Context) { + if m == nil { + return + } + m.finalizerRemoveFailure.Add(ctx, 1) +} + +func (m *Metrics) RecordStatusUpdateDuration(ctx context.Context, d time.Duration) { + if m == nil { + return + } + m.statusUpdateDuration.Record(ctx, d.Seconds()) +} + +func (m *Metrics) IncStatusUpdateFailure(ctx context.Context) { + if m == nil { + return + } + m.statusUpdateFailure.Add(ctx, 1) +} diff --git a/pkg/telemetry/metrics_test.go b/pkg/telemetry/metrics_test.go new file mode 100644 index 0000000..ff8a90f --- /dev/null +++ b/pkg/telemetry/metrics_test.go @@ -0,0 +1,117 @@ +package telemetry + +import ( + "context" + "testing" + "time" + + "github.com/krateoplatformops/provider-runtime/pkg/logging" + "go.opentelemetry.io/otel/sdk/metric" + metricdata "go.opentelemetry.io/otel/sdk/metric/metricdata" +) + +func TestSetupDisabledReturnsNoopShutdown(t *testing.T) { + metrics, shutdown, err := Setup(context.Background(), logging.NewNopLogger(), Config{}) + if err != nil { + t.Fatalf("Setup() returned error: %v", err) + } + if metrics != nil { + t.Fatalf("Setup() metrics = %#v, want nil", metrics) + } + if shutdown == nil { + t.Fatal("Setup() shutdown = nil, want no-op shutdown") + } + if err := shutdown(context.Background()); err != nil { + t.Fatalf("shutdown() returned error: %v", err) + } +} + +func TestNewMetricsRecordsData(t *testing.T) { + reader := metric.NewManualReader() + provider := metric.NewMeterProvider(metric.WithReader(reader)) + ctx := context.Background() + t.Cleanup(func() { + if err := provider.Shutdown(ctx); err != nil { + t.Fatalf("provider.Shutdown() returned error: %v", err) + } + }) + + metrics, err := newMetrics(provider.Meter("github.com/krateoplatformops/provider-runtime/test")) + if err != nil { + t.Fatalf("newMetrics() returned error: %v", err) + } + + metrics.IncStartupSuccess(ctx) + metrics.RecordGetDuration(ctx, 25*time.Millisecond) + metrics.IncGetFailure(ctx) + metrics.RecordReconcileDuration(ctx, 150*time.Millisecond) + metrics.AddQueueDepth(ctx, 1) + metrics.RecordQueueWaitDuration(ctx, 75*time.Millisecond) + metrics.RecordQueueOldestItemAge(ctx, 5*time.Second) + metrics.RecordQueueWorkDuration(ctx, 125*time.Millisecond) + metrics.IncReconcileRequeueAfter(ctx) + metrics.IncReconcileRequeueImmediate(ctx) + metrics.IncReconcileErrorRequeue(ctx) + metrics.AddReconcileInFlight(1) + metrics.IncReconcileFailure(ctx) + + var rm metricdata.ResourceMetrics + if err := reader.Collect(ctx, &rm); err != nil { + t.Fatalf("reader.Collect() returned error: %v", err) + } + + if !hasMetric(rm, "provider_runtime.startup.success") { + t.Fatal("expected startup success metric to be collected") + } + if !hasMetric(rm, "provider_runtime.reconcile.get.duration_seconds") { + t.Fatal("expected get duration metric to be collected") + } + if !hasMetric(rm, "provider_runtime.reconcile.get.failure") { + t.Fatal("expected get failure metric to be collected") + } + if !hasMetric(rm, "provider_runtime.reconcile.duration_seconds") { + t.Fatal("expected reconcile duration metric to be collected") + } + if !hasMetric(rm, "provider_runtime.reconcile.queue.depth") { + t.Fatal("expected queue depth metric to be collected") + } + if !hasMetric(rm, "provider_runtime.reconcile.queue.wait.duration_seconds") { + t.Fatal("expected queue wait metric to be collected") + } + if !hasMetric(rm, "provider_runtime.reconcile.queue.oldest_item_age_seconds") { + t.Fatal("expected queue oldest item age metric to be collected") + } + if !hasMetric(rm, "provider_runtime.reconcile.queue.work.duration_seconds") { + t.Fatal("expected queue work duration metric to be collected") + } + if !hasMetric(rm, "provider_runtime.reconcile.queue.requeues") { + t.Fatal("expected queue requeue total metric to be collected") + } + if !hasMetric(rm, "provider_runtime.reconcile.requeue.after") { + t.Fatal("expected reconcile requeue-after metric to be collected") + } + if !hasMetric(rm, "provider_runtime.reconcile.requeue.immediate") { + t.Fatal("expected reconcile requeue-immediate metric to be collected") + } + if !hasMetric(rm, "provider_runtime.reconcile.requeue.error") { + t.Fatal("expected reconcile requeue-error metric to be collected") + } + if !hasMetric(rm, "provider_runtime.reconcile.in_flight") { + t.Fatal("expected reconcile in-flight metric to be collected") + } + if !hasMetric(rm, "provider_runtime.reconcile.failure") { + t.Fatal("expected reconcile failure metric to be collected") + } +} + +func hasMetric(rm metricdata.ResourceMetrics, name string) bool { + for _, scope := range rm.ScopeMetrics { + for _, metric := range scope.Metrics { + if metric.Name == name { + return true + } + } + } + + return false +} From 093be930d278f1d6f0ece3c9778326e46ffd93bd Mon Sep 17 00:00:00 2001 From: Matteo Gastaldello Date: Wed, 15 Apr 2026 10:11:20 +0200 Subject: [PATCH 2/4] fix(reconciler): update reconcile to handle errors as expected by controller-runtime --- pkg/reconciler/reconciler.go | 59 ++++++++++++++++++------------------ 1 file changed, 30 insertions(+), 29 deletions(-) diff --git a/pkg/reconciler/reconciler.go b/pkg/reconciler/reconciler.go index 87d90b3..b75477d 100644 --- a/pkg/reconciler/reconciler.go +++ b/pkg/reconciler/reconciler.go @@ -746,9 +746,9 @@ func (r *Reconciler) Reconcile(ctx context.Context, req reconcile.Request) (resu } managed.SetConditions(prv1.Deleting(), prv1.ReconcileError(err)) if err := statusUpdate(func() error { return r.client.Status().Update(ctx, managed) }); err != nil { - return reconcile.Result{Requeue: true}, errors.Wrap(err, errUpdateManagedStatus) + return reconcile.Result{}, errors.Wrap(err, errUpdateManagedStatus) } - return reconcile.Result{Requeue: true}, nil + return reconcile.Result{}, errors.Wrap(err, errUpdateManagedStatus) } // We've successfully unpublished our managed resource's connection @@ -768,7 +768,7 @@ func (r *Reconciler) Reconcile(ctx context.Context, req reconcile.Request) (resu throttledRecorder.Event(managed, event.Warning(reasonCannotInitialize, actionProcessEvent, errors.New(errCreateIncomplete))) managed.SetConditions(prv1.Creating(), prv1.ReconcileError(errors.New(errCreateIncomplete))) if err := statusUpdate(func() error { return r.client.Status().Update(ctx, managed) }); err != nil { - return reconcile.Result{Requeue: false}, errors.Wrap(err, errUpdateManagedStatus) + return reconcile.Result{}, errors.Wrap(err, errUpdateManagedStatus) } return reconcile.Result{Requeue: false}, nil } @@ -789,9 +789,9 @@ func (r *Reconciler) Reconcile(ctx context.Context, req reconcile.Request) (resu throttledRecorder.Event(managed, event.Warning(reasonCannotConnect, actionProcessEvent, err)) managed.SetConditions(prv1.ReconcileError(errors.Wrap(err, errReconcileConnect))) if err := statusUpdate(func() error { return r.client.Status().Update(ctx, managed) }); err != nil { - return reconcile.Result{Requeue: true}, errors.Wrap(err, errUpdateManagedStatus) + return reconcile.Result{}, errors.Wrap(err, errUpdateManagedStatus) } - return reconcile.Result{Requeue: true}, nil + return reconcile.Result{}, errors.Wrap(err, errReconcileConnect) } defer func() { if err := r.external.Disconnect(ctx); err != nil { @@ -815,11 +815,12 @@ func (r *Reconciler) Reconcile(ctx context.Context, req reconcile.Request) (resu return reconcile.Result{Requeue: true}, nil } throttledRecorder.Event(managed, event.Warning(reasonCannotObserve, actionObserveExternalResource, err)) - managed.SetConditions(prv1.ReconcileError(errors.Wrap(err, errReconcileObserve))) + werr := errors.Wrap(err, errReconcileObserve) + managed.SetConditions(prv1.ReconcileError(werr)) if err := statusUpdate(func() error { return r.client.Status().Update(ctx, managed) }); err != nil { - return reconcile.Result{Requeue: true}, errors.Wrap(err, errUpdateManagedStatus) + return reconcile.Result{}, errors.Wrap(err, errUpdateManagedStatus) } - return reconcile.Result{Requeue: true}, nil + return reconcile.Result{}, werr } // In the observe-only mode, !observation.ResourceExists will be an error @@ -828,9 +829,9 @@ func (r *Reconciler) Reconcile(ctx context.Context, req reconcile.Request) (resu throttledRecorder.Event(managed, event.Warning(reasonCannotObserve, actionObserveExternalResource, errors.New(errExternalResourceNotExist))) managed.SetConditions(prv1.ReconcileError(errors.Wrap(errors.New(errExternalResourceNotExist), errReconcileObserve))) if err := statusUpdate(func() error { return r.client.Status().Update(ctx, managed) }); err != nil { - return reconcile.Result{Requeue: true}, errors.Wrap(err, errUpdateManagedStatus) + return reconcile.Result{}, errors.Wrap(err, errUpdateManagedStatus) } - return reconcile.Result{Requeue: true}, nil + return reconcile.Result{}, errors.Wrap(errors.New(errExternalResourceNotExist), errReconcileObserve) } // If this resource has a non-zero creation grace period we want to wait @@ -864,9 +865,9 @@ func (r *Reconciler) Reconcile(ctx context.Context, req reconcile.Request) (resu throttledRecorder.Event(managed, event.Warning(reasonCannotDelete, actionDeleteExternalResource, err)) managed.SetConditions(prv1.Deleting(), prv1.ReconcileError(errors.Wrap(err, errReconcileDelete))) if err := statusUpdate(func() error { return r.client.Status().Update(ctx, managed) }); err != nil { - return reconcile.Result{Requeue: true}, errors.Wrap(err, errUpdateManagedStatus) + return reconcile.Result{}, errors.Wrap(err, errUpdateManagedStatus) } - return reconcile.Result{Requeue: true}, nil + return reconcile.Result{}, errors.Wrap(err, errReconcileDelete) } // We've successfully requested deletion of our external resource. @@ -899,9 +900,9 @@ func (r *Reconciler) Reconcile(ctx context.Context, req reconcile.Request) (resu } managed.SetConditions(prv1.Deleting(), prv1.ReconcileError(err)) if err := statusUpdate(func() error { return r.client.Status().Update(ctx, managed) }); err != nil { - return reconcile.Result{Requeue: true}, errors.Wrap(err, errUpdateManagedStatus) + return reconcile.Result{}, errors.Wrap(err, errUpdateManagedStatus) } - return reconcile.Result{Requeue: true}, nil + return reconcile.Result{}, errors.Wrap(err, errUpdateManagedStatus) } // We've successfully deleted our external resource (if necessary) and @@ -925,9 +926,9 @@ func (r *Reconciler) Reconcile(ctx context.Context, req reconcile.Request) (resu } managed.SetConditions(prv1.ReconcileError(err)) if err := statusUpdate(func() error { return r.client.Status().Update(ctx, managed) }); err != nil { - return reconcile.Result{Requeue: true}, errors.Wrap(err, errUpdateManagedStatus) + return reconcile.Result{}, errors.Wrap(err, errUpdateManagedStatus) } - return reconcile.Result{Requeue: true}, nil + return reconcile.Result{}, errors.Wrap(err, errCreateIncomplete) } if !observation.ResourceExists && meta.ShouldCreate(managed) { @@ -947,9 +948,9 @@ func (r *Reconciler) Reconcile(ctx context.Context, req reconcile.Request) (resu throttledRecorder.Event(managed, event.Warning(reasonCannotUpdateManaged, actionUpdateExternalResource, errors.Wrap(err, errUpdateManaged))) managed.SetConditions(prv1.Creating(), prv1.ReconcileError(errors.Wrap(err, errUpdateManaged))) if err := statusUpdate(func() error { return r.client.Status().Update(ctx, managed) }); err != nil { - return reconcile.Result{Requeue: true}, errors.Wrap(err, errUpdateManagedStatus) + return reconcile.Result{}, errors.Wrap(err, errUpdateManagedStatus) } - return reconcile.Result{Requeue: true}, nil + return reconcile.Result{}, errors.Wrap(err, errUpdateManaged) } started = time.Now() @@ -987,9 +988,9 @@ func (r *Reconciler) Reconcile(ctx context.Context, req reconcile.Request) (resu managed.SetConditions(prv1.Creating(), prv1.ReconcileError(errors.Wrap(err, errReconcileCreate))) if err := statusUpdate(func() error { return r.client.Status().Update(ctx, managed) }); err != nil { - return reconcile.Result{Requeue: true}, errors.Wrap(err, errUpdateManagedStatus) + return reconcile.Result{}, errors.Wrap(err, errUpdateManagedStatus) } - return reconcile.Result{Requeue: true}, nil + return reconcile.Result{}, errors.Wrap(err, errReconcileCreate) } // In some cases our external-name may be set by Create above. @@ -1012,9 +1013,9 @@ func (r *Reconciler) Reconcile(ctx context.Context, req reconcile.Request) (resu throttledRecorder.Event(managed, event.Warning(reasonCannotUpdateManaged, actionUpdateExternalResource, errors.Wrap(err, errUpdateManagedAnnotations))) managed.SetConditions(prv1.Creating(), prv1.ReconcileError(errors.Wrap(err, errUpdateManagedAnnotations))) if err := statusUpdate(func() error { return r.client.Status().Update(ctx, managed) }); err != nil { - return reconcile.Result{Requeue: true}, errors.Wrap(err, errUpdateManagedStatus) + return reconcile.Result{}, errors.Wrap(err, errUpdateManagedStatus) } - return reconcile.Result{Requeue: true}, nil + return reconcile.Result{}, errors.Wrap(err, errUpdateManagedAnnotations) } // We've successfully created our external resource. In many cases the @@ -1043,9 +1044,9 @@ func (r *Reconciler) Reconcile(ctx context.Context, req reconcile.Request) (resu throttledRecorder.Event(managed, event.Warning(reasonCannotUpdateManaged, actionUpdateExternalResource, err)) managed.SetConditions(prv1.ReconcileError(errors.Wrap(err, errUpdateManaged))) if err := statusUpdate(func() error { return r.client.Status().Update(ctx, managed) }); err != nil { - return reconcile.Result{Requeue: true}, errors.Wrap(err, errUpdateManagedStatus) + return reconcile.Result{}, errors.Wrap(err, errUpdateManagedStatus) } - return reconcile.Result{Requeue: true}, nil + return reconcile.Result{}, errors.Wrap(err, errUpdateManaged) } } @@ -1060,7 +1061,7 @@ func (r *Reconciler) Reconcile(ctx context.Context, req reconcile.Request) (resu log.Info("External resource is up to date", "requeue-after", time.Now().Add(reconcileAfter)) managed.SetConditions(prv1.ReconcileSuccess()) if err := statusUpdate(func() error { return r.client.Status().Update(ctx, managed) }); err != nil { - return reconcile.Result{RequeueAfter: reconcileAfter}, errors.Wrap(err, errUpdateManagedStatus) + return reconcile.Result{}, errors.Wrap(err, errUpdateManagedStatus) } return reconcile.Result{RequeueAfter: reconcileAfter}, nil } @@ -1075,7 +1076,7 @@ func (r *Reconciler) Reconcile(ctx context.Context, req reconcile.Request) (resu log.Debug("Skipping update due to managementPolicies. Reconciliation succeeded", "requeue-after", time.Now().Add(reconcileAfter)) managed.SetConditions(prv1.ReconcileSuccess()) if err := statusUpdate(func() error { return r.client.Status().Update(ctx, managed) }); err != nil { - return reconcile.Result{RequeueAfter: reconcileAfter}, errors.Wrap(err, errUpdateManagedStatus) + return reconcile.Result{}, errors.Wrap(err, errUpdateManagedStatus) } return reconcile.Result{RequeueAfter: reconcileAfter}, nil } @@ -1093,9 +1094,9 @@ func (r *Reconciler) Reconcile(ctx context.Context, req reconcile.Request) (resu throttledRecorder.Event(managed, event.Warning(reasonCannotUpdate, actionUpdateExternalResource, err)) managed.SetConditions(prv1.ReconcileError(errors.Wrap(err, errReconcileUpdate))) if err := statusUpdate(func() error { return r.client.Status().Update(ctx, managed) }); err != nil { - return reconcile.Result{Requeue: true}, errors.Wrap(err, errUpdateManagedStatus) + return reconcile.Result{}, errors.Wrap(err, errUpdateManagedStatus) } - return reconcile.Result{Requeue: true}, nil + return reconcile.Result{}, errors.Wrap(err, errReconcileUpdate) } // We've successfully updated our external resource. Per the below issue @@ -1108,7 +1109,7 @@ func (r *Reconciler) Reconcile(ctx context.Context, req reconcile.Request) (resu record.Event(managed, event.Normal(reasonUpdated, actionUpdateEvent, "Successfully requested update of external resource")) managed.SetConditions(prv1.ReconcileSuccess()) if err := statusUpdate(func() error { return r.client.Status().Update(ctx, managed) }); err != nil { - return reconcile.Result{RequeueAfter: reconcileAfter}, errors.Wrap(err, errUpdateManagedStatus) + return reconcile.Result{}, errors.Wrap(err, errUpdateManagedStatus) } return reconcile.Result{RequeueAfter: reconcileAfter}, nil } From 91b1ba23cb7268feee8db720f6f2afdb1f8cd059 Mon Sep 17 00:00:00 2001 From: Matteo Gastaldello Date: Wed, 15 Apr 2026 15:04:33 +0200 Subject: [PATCH 3/4] fix(reconciler): update reconcile to return empty result on status update error --- pkg/reconciler/reconciler.go | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/pkg/reconciler/reconciler.go b/pkg/reconciler/reconciler.go index b75477d..29b26cb 100644 --- a/pkg/reconciler/reconciler.go +++ b/pkg/reconciler/reconciler.go @@ -881,7 +881,7 @@ func (r *Reconciler) Reconcile(ctx context.Context, req reconcile.Request) (resu record.Event(managed, event.Normal(reasonDeleted, actionDeleteEvent, "Successfully requested deletion of external resource")) managed.SetConditions(prv1.Deleting(), prv1.ReconcileSuccess()) if err := statusUpdate(func() error { return r.client.Status().Update(ctx, managed) }); err != nil { - return reconcile.Result{Requeue: true}, errors.Wrap(err, errUpdateManagedStatus) + return reconcile.Result{}, errors.Wrap(err, errUpdateManagedStatus) } return reconcile.Result{Requeue: true}, nil } @@ -1026,7 +1026,7 @@ func (r *Reconciler) Reconcile(ctx context.Context, req reconcile.Request) (resu record.Event(managed, event.Normal(reasonCreated, actionCreateEvent, "Successfully requested creation of external resource")) managed.SetConditions(prv1.Creating(), prv1.ReconcileSuccess()) if err := statusUpdate(func() error { return r.client.Status().Update(ctx, managed) }); err != nil { - return reconcile.Result{Requeue: true}, errors.Wrap(err, errUpdateManagedStatus) + return reconcile.Result{}, errors.Wrap(err, errUpdateManagedStatus) } return reconcile.Result{Requeue: true}, nil } From 1f9ed9111c206fbeaf4c4437df4343a791ade7d8 Mon Sep 17 00:00:00 2001 From: Matteo Gastaldello Date: Wed, 15 Apr 2026 16:11:07 +0200 Subject: [PATCH 4/4] fix(tests): update reconciler test expectations to handle error wrapping --- pkg/reconciler/reconciler_test.go | 24 ++++++++++++------------ 1 file changed, 12 insertions(+), 12 deletions(-) diff --git a/pkg/reconciler/reconciler_test.go b/pkg/reconciler/reconciler_test.go index ab5f413..9c7b0c8 100644 --- a/pkg/reconciler/reconciler_test.go +++ b/pkg/reconciler/reconciler_test.go @@ -103,7 +103,7 @@ func TestReconciler(t *testing.T) { WithFinalizer(resource.FinalizerFns{RemoveFinalizerFn: func(_ context.Context, _ resource.Object) error { return errBoom }}), }, }, - want: want{result: reconcile.Result{Requeue: true}}, + want: want{err: errors.Wrap(errBoom, errUpdateManagedStatus)}, }, "DeleteSuccessfulDeletionPolicyOrphan": { reason: "Successful managed resource deletion with deletion policy Orphan should not trigger a requeue or status update.", @@ -179,7 +179,7 @@ func TestReconciler(t *testing.T) { })), }, }, - want: want{result: reconcile.Result{Requeue: true}}, + want: want{err: errors.Wrap(errBoom, errReconcileConnect)}, }, "ExternalDisconnectError": { reason: "Error disconnecting from the provider should not trigger requeue.", @@ -248,7 +248,7 @@ func TestReconciler(t *testing.T) { })), }, }, - want: want{result: reconcile.Result{Requeue: true}}, + want: want{err: errors.Wrap(errBoom, errReconcileObserve)}, }, "CreationGracePeriod": { reason: "If our resource appears not to exist during the creation grace period we should return early.", @@ -322,7 +322,7 @@ func TestReconciler(t *testing.T) { })), }, }, - want: want{result: reconcile.Result{Requeue: true}}, + want: want{err: errors.Wrap(errBoom, errReconcileDelete)}, }, "ExternalDeleteSuccessful": { reason: "A deleted managed resource with the 'delete' reclaim policy should delete its external resource then requeue after a short wait.", @@ -414,7 +414,7 @@ func TestReconciler(t *testing.T) { WithFinalizer(resource.FinalizerFns{RemoveFinalizerFn: func(_ context.Context, _ resource.Object) error { return errBoom }}), }, }, - want: want{result: reconcile.Result{Requeue: true}}, + want: want{err: errors.Wrap(errBoom, errUpdateManagedStatus)}, }, "DeleteSuccessfulDeletionPolicyDelete": { reason: "Successful managed resource deletion with deletion policy Delete should not trigger a requeue or status update.", @@ -471,7 +471,7 @@ func TestReconciler(t *testing.T) { WithFinalizer(resource.FinalizerFns{AddFinalizerFn: func(_ context.Context, _ resource.Object) error { return errBoom }}), }, }, - want: want{result: reconcile.Result{Requeue: true}}, + want: want{err: errors.Wrap(errBoom, errCreateIncomplete)}, }, "UpdateCreatePendingError": { reason: "Errors while updating our external-create-pending annotation should trigger a requeue after a short wait.", @@ -509,7 +509,7 @@ func TestReconciler(t *testing.T) { WithFinalizer(resource.FinalizerFns{AddFinalizerFn: func(_ context.Context, _ resource.Object) error { return nil }}), }, }, - want: want{result: reconcile.Result{Requeue: true}}, + want: want{err: errors.Wrap(errBoom, errUpdateManaged)}, }, "CreateExternalError": { reason: "Errors while creating an external resource should trigger a requeue after a short wait.", @@ -552,7 +552,7 @@ func TestReconciler(t *testing.T) { WithFinalizer(resource.FinalizerFns{AddFinalizerFn: func(_ context.Context, _ resource.Object) error { return nil }}), }, }, - want: want{result: reconcile.Result{Requeue: true}}, + want: want{err: errors.Wrap(errBoom, errReconcileCreate)}, }, "UpdateCriticalAnnotationsError": { reason: "Errors updating critical annotations after creation should trigger a requeue after a short wait.", @@ -593,7 +593,7 @@ func TestReconciler(t *testing.T) { WithCriticalAnnotationUpdater(CriticalAnnotationUpdateFn(func(ctx context.Context, o client.Object) error { return errBoom })), }, }, - want: want{result: reconcile.Result{Requeue: true}}, + want: want{err: errors.Wrap(errBoom, errUpdateManagedAnnotations)}, }, "CreateSuccessful": { reason: "Successful managed resource creation should trigger a requeue after a short wait.", @@ -658,7 +658,7 @@ func TestReconciler(t *testing.T) { WithFinalizer(resource.FinalizerFns{AddFinalizerFn: func(_ context.Context, _ resource.Object) error { return nil }}), }, }, - want: want{result: reconcile.Result{Requeue: true}}, + want: want{err: errors.Wrap(errBoom, errUpdateManaged)}, }, "ExternalResourceUpToDate": { reason: "When the external resource exists and is up to date a requeue should be triggered after a long wait.", @@ -727,7 +727,7 @@ func TestReconciler(t *testing.T) { WithFinalizer(resource.FinalizerFns{AddFinalizerFn: func(_ context.Context, _ resource.Object) error { return nil }}), }, }, - want: want{result: reconcile.Result{Requeue: true}}, + want: want{err: errors.Wrap(errBoom, errReconcileUpdate)}, }, "UpdateSuccessful": { reason: "A successful managed resource update should trigger a requeue after a long wait.", @@ -962,7 +962,7 @@ func TestReconciler(t *testing.T) { })), }, }, - want: want{result: reconcile.Result{Requeue: true}}, + want: want{err: errors.Wrap(errors.New(errExternalResourceNotExist), errReconcileObserve)}, }, }