From d6768bd2c1a9200083fa2cc4959d4de23551d2dd Mon Sep 17 00:00:00 2001 From: Matteo Gastaldello Date: Wed, 15 Apr 2026 17:15:59 +0200 Subject: [PATCH 1/4] feat integrate OpenTelemetry metrics for queue and reconciliation tracking - Added telemetry metrics to monitor queue depth, wait duration, and work duration. - Implemented metrics for reconciliation success, failure, and requeue counts. - Introduced InstrumentedQueue to wrap the priority queue and record metrics. - Enhanced controller worker to track processing durations and reconcile metrics. - Created a new telemetry package with metrics setup and recording functionality. - Added tests for metrics setup and validation of recorded data. --- go.mod | 49 ++- go.sum | 97 +++-- pkg/controller/builder/builder.go | 1 + pkg/controller/builder/options.go | 8 + pkg/controller/controller.go | 12 +- pkg/controller/queue_instrumentation.go | 185 ++++++++++ pkg/controller/worker.go | 68 +++- pkg/telemetry/metrics.go | 462 ++++++++++++++++++++++++ pkg/telemetry/metrics_test.go | 130 +++++++ 9 files changed, 958 insertions(+), 54 deletions(-) create mode 100644 pkg/controller/queue_instrumentation.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 bdffaeb..b878c03 100644 --- a/go.mod +++ b/go.mod @@ -1,6 +1,6 @@ module github.com/krateoplatformops/unstructured-runtime -go 1.25.3 +go 1.25.6 require ( github.com/fsnotify/fsnotify v1.9.0 @@ -9,20 +9,31 @@ require ( github.com/google/btree v1.1.3 github.com/google/go-cmp v0.7.0 github.com/google/gofuzz v1.2.0 - github.com/krateoplatformops/plumbing v1.3.1 + github.com/krateoplatformops/plumbing v1.7.2 github.com/onsi/ginkgo/v2 v2.27.2 github.com/onsi/gomega v1.38.2 github.com/prometheus/client_golang v1.23.2 github.com/stretchr/testify v1.11.1 github.com/twmb/murmur3 v1.1.8 golang.org/x/time v0.12.0 - k8s.io/api v0.35.2 - k8s.io/apimachinery v0.35.2 - k8s.io/client-go v0.35.2 + k8s.io/api v0.35.3 + k8s.io/apimachinery v0.35.3 + k8s.io/client-go v0.35.3 k8s.io/utils v0.0.0-20251002143259-bc988d571ff4 sigs.k8s.io/e2e-framework v0.6.0 ) +require ( + github.com/cenkalti/backoff/v5 v5.0.3 // indirect + github.com/go-logr/stdr v1.2.2 // indirect + github.com/grpc-ecosystem/grpc-gateway/v2 v2.28.0 // indirect + go.opentelemetry.io/auto/sdk v1.2.1 // indirect + go.opentelemetry.io/proto/otlp v1.10.0 // 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 +) + require ( github.com/Masterminds/semver/v3 v3.4.0 // indirect github.com/beorn7/perks v1.0.1 // indirect @@ -57,23 +68,27 @@ require ( github.com/spf13/pflag v1.0.10 // indirect github.com/vladimirvivien/gexe v0.4.1 // indirect github.com/x448/float16 v0.8.4 // indirect - go.opentelemetry.io/otel v1.37.0 // indirect - go.opentelemetry.io/otel/trace v1.37.0 // indirect + 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 + go.opentelemetry.io/otel/trace v1.43.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 - google.golang.org/protobuf v1.36.8 // 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 + 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.v3 v3.0.1 // indirect - k8s.io/component-base v0.35.0 // indirect + k8s.io/component-base v0.35.1 // indirect k8s.io/klog/v2 v2.130.1 // indirect k8s.io/kube-openapi v0.0.0-20250910181357-589584f1c912 // indirect sigs.k8s.io/controller-runtime v0.22.3 // indirect diff --git a/go.sum b/go.sum index 5083f07..9a8adb6 100644 --- a/go.sum +++ b/go.sum @@ -6,6 +6,8 @@ github.com/beorn7/perks v1.0.1 h1:VlbKKnNfV8bJzeqoa4cOKqO6bYr3WgKZxO8Z16+hsOM= github.com/beorn7/perks v1.0.1/go.mod h1:G2ZrVWU2WbWT9wwq4/hrbKbnv/1ERSJQ0ibhJ6rlkpw= 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/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/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= @@ -26,8 +28,11 @@ github.com/gkampitakis/go-diff v1.3.2 h1:Qyn0J9XJSDTgnsgHRdz9Zp24RaJeKMUHg2+PDZZ github.com/gkampitakis/go-diff v1.3.2/go.mod h1:LLgOrpqleQe26cte8s36HTWcTmMEur6OPYerdAAS9tk= github.com/gkampitakis/go-snaps v0.5.15 h1:amyJrvM1D33cPHwVrjo9jQxX8g/7E2wYdZ+01KS3zGE= github.com/gkampitakis/go-snaps v0.5.15/go.mod h1:HNpx/9GoKisdhw9AFOBT1N7DBs9DiHo/hGheFGBZ+mc= +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= +github.com/go-logr/stdr v1.2.2/go.mod h1:mMo/vtBO5dYbehREoey6XUKy/eSumjCCveDpRre4VKE= github.com/go-logr/zapr v1.3.0 h1:XGdV8XW8zdwFiwOA2Dryh1gj2KRQyOOoNmBy4EplIcQ= github.com/go-logr/zapr v1.3.0/go.mod h1:YKepepNBd1u/oyhd/yQmtjVXmm9uML4IXUgMOwR8/Gg= github.com/go-openapi/jsonpointer v0.21.1 h1:whnzv/pNXtK2FbX/W9yJfRmE2gsmkfahjMKB0fZvcic= @@ -44,6 +49,8 @@ github.com/goccy/go-yaml v1.18.0 h1:8W7wMFS12Pcas7KU+VVkaiCng+kG8QiFeFwzFb+rwuw= github.com/goccy/go-yaml v1.18.0/go.mod h1:XBurs7gK8ATbW4ZPGKgcbrY1Br56PdM69F7LkFRi1kA= github.com/golang-jwt/jwt/v5 v5.2.2 h1:Rl4B7itRWVtYIHFrSNd7vhTiz9UpLdi6gZhZ3wEeDy8= github.com/golang-jwt/jwt/v5 v5.2.2/go.mod h1:pqrtFR0X4osieyHYxtmOUWsAWrfe1Q5UVIyoH402zdk= +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/gnostic-models v0.7.0 h1:qwTtogB15McXDaNqTZdzPJRHvaVJlAl+HVQnLmJEJxo= @@ -59,6 +66,8 @@ 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/gorilla/websocket v1.5.4-0.20250319132907-e064f32e3674 h1:JeSE6pjso5THxAzdVpqr6/geYxZytqFMBCOtn/ujyeo= github.com/gorilla/websocket v1.5.4-0.20250319132907-e064f32e3674/go.mod h1:r4w70xmWCQKmi1ONH4KIaBptdivuRPyosB9RmPlGEwA= +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/josharian/intern v1.0.0 h1:vlS4z54oSdjm0bgjRigI+G1HpF+tI+9rE5LLzOg8HmY= github.com/josharian/intern v1.0.0/go.mod h1:5DoeVV0s6jJacbCEi61lwdGj/aVlrQvzHFFd8Hwg//Y= github.com/joshdk/go-junit v1.0.0 h1:S86cUKIdwBHWwA6xCmFlf3RTLfVXYQfvanM5Uh+K6GE= @@ -71,8 +80,8 @@ github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE= github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk= github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY= github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE= -github.com/krateoplatformops/plumbing v1.3.1 h1:DQc/ohBiGfI6mC3l2T+3ZvtSR0r3hOLqdxu73xjZUDM= -github.com/krateoplatformops/plumbing v1.3.1/go.mod h1:51a1TKAoCXRHH7cJB8kiYZcIV8ci3PLPtabqBdoD9Zg= +github.com/krateoplatformops/plumbing v1.7.2 h1:I9rvpKPQ1eswRZkdwCT+xfdDRLODSsL1c7a6g8K3S9c= +github.com/krateoplatformops/plumbing v1.7.2/go.mod h1:UrYvAvnfacOiOLlQbIg6J08zYC+XJ9ZMKbecQq+4jV4= github.com/kylelemons/godebug v1.1.0 h1:RPNrshWIDI6G2gRW9EHilWtl7Z6Sb1BR0xunSBf0SNc= github.com/kylelemons/godebug v1.1.0/go.mod h1:9/0rRGxNHcop5bhtWyNeEfOS8JIWk580+fNqagV/RAw= github.com/mailru/easyjson v0.9.0 h1:PrnmzHw7262yW8sTBwxi1PdJA3Iw/EKBa8psRf7d9a4= @@ -137,10 +146,22 @@ github.com/vladimirvivien/gexe v0.4.1 h1:W9gWkp8vSPjDoXDu04Yp4KljpVMaSt8IQuHswLD github.com/vladimirvivien/gexe v0.4.1/go.mod h1:3gjgTqE2c0VyHnU5UOIwk7gyNzZDGulPb/DJPgcw64E= 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/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/trace v1.37.0 h1:HLdcFNbRQBE2imdSEgm/kwqmQj1Or1l/7bW6mxVK7z4= -go.opentelemetry.io/otel/trace v1.37.0/go.mod h1:TlgrlQ+PtQO5XFerSPUYG0JSgGyryXewPGyayAWSBS0= +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/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/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= @@ -151,26 +172,34 @@ go.yaml.in/yaml/v2 v2.4.3 h1:6gvOSjQoTB3vt1l+CU+tSyi/HOjfOjRLJ4YwYZGwRO0= go.yaml.in/yaml/v2 v2.4.3/go.mod h1:zSxWcmIDjOzPXpjlTTbAsKokqkDNAVtZO0WOMiT90s8= 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/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/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/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.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= -google.golang.org/protobuf v1.36.8 h1:xHScyCOEuuwZEc6UtSOvPbAT4zRh0xcNRYekJwfqyMc= -google.golang.org/protobuf v1.36.8/go.mod h1:fuxRtAxBytpl4zzqUh6/eyUujkJdNiuEkXntxiD/uRU= +golang.org/x/tools v0.42.0 h1:uNgphsn75Tdz5Ji2q36v/nsFSfR/9BRFvqhGBaJGd5k= +golang.org/x/tools v0.42.0/go.mod h1:Ma6lCIwGZvHK6XtgbswSoWroEkhugApmsXyrUmBhfr0= +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= @@ -181,16 +210,16 @@ gopkg.in/inf.v0 v0.9.1/go.mod h1:cWUDdTG/fYaXco+Dcufb5Vnc6Gp2YChqWtbxRZE0mXw= gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA= gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM= -k8s.io/api v0.35.2 h1:tW7mWc2RpxW7HS4CoRXhtYHSzme1PN1UjGHJ1bdrtdw= -k8s.io/api v0.35.2/go.mod h1:7AJfqGoAZcwSFhOjcGM7WV05QxMMgUaChNfLTXDRE60= -k8s.io/apiextensions-apiserver v0.35.0 h1:3xHk2rTOdWXXJM+RDQZJvdx0yEOgC0FgQ1PlJatA5T4= -k8s.io/apiextensions-apiserver v0.35.0/go.mod h1:E1Ahk9SADaLQ4qtzYFkwUqusXTcaV2uw3l14aqpL2LU= -k8s.io/apimachinery v0.35.2 h1:NqsM/mmZA7sHW02JZ9RTtk3wInRgbVxL8MPfzSANAK8= -k8s.io/apimachinery v0.35.2/go.mod h1:jQCgFZFR1F4Ik7hvr2g84RTJSZegBc8yHgFWKn//hns= -k8s.io/client-go v0.35.2 h1:YUfPefdGJA4aljDdayAXkc98DnPkIetMl4PrKX97W9o= -k8s.io/client-go v0.35.2/go.mod h1:4QqEwh4oQpeK8AaefZ0jwTFJw/9kIjdQi0jpKeYvz7g= -k8s.io/component-base v0.35.0 h1:+yBrOhzri2S1BVqyVSvcM3PtPyx5GUxCK2tinZz1G94= -k8s.io/component-base v0.35.0/go.mod h1:85SCX4UCa6SCFt6p3IKAPej7jSnF3L8EbfSyMZayJR0= +k8s.io/api v0.35.3 h1:pA2fiBc6+N9PDf7SAiluKGEBuScsTzd2uYBkA5RzNWQ= +k8s.io/api v0.35.3/go.mod h1:9Y9tkBcFwKNq2sxwZTQh1Njh9qHl81D0As56tu42GA4= +k8s.io/apiextensions-apiserver v0.35.1 h1:p5vvALkknlOcAqARwjS20kJffgzHqwyQRM8vHLwgU7w= +k8s.io/apiextensions-apiserver v0.35.1/go.mod h1:2CN4fe1GZ3HMe4wBr25qXyJnJyZaquy4nNlNmb3R7AQ= +k8s.io/apimachinery v0.35.3 h1:MeaUwQCV3tjKP4bcwWGgZ/cp/vpsRnQzqO6J6tJyoF8= +k8s.io/apimachinery v0.35.3/go.mod h1:jQCgFZFR1F4Ik7hvr2g84RTJSZegBc8yHgFWKn//hns= +k8s.io/client-go v0.35.3 h1:s1lZbpN4uI6IxeTM2cpdtrwHcSOBML1ODNTCCfsP1pg= +k8s.io/client-go v0.35.3/go.mod h1:RzoXkc0mzpWIDvBrRnD+VlfXP+lRzqQjCmKtiwZ8Q9c= +k8s.io/component-base v0.35.1 h1:XgvpRf4srp037QWfGBLFsYMUQJkE5yMa94UsJU7pmcE= +k8s.io/component-base v0.35.1/go.mod h1:HI/6jXlwkiOL5zL9bqA3en1Ygv60F03oEpnuU1G56Bs= k8s.io/klog/v2 v2.130.1 h1:n9Xl7H1Xvksem4KFG4PYbdQCQxqc/tTUyrgXaOhHSzk= k8s.io/klog/v2 v2.130.1/go.mod h1:3Jpz1GvMt720eyJH1ckRHK1EDfpxISzJ7I9OYgaDtPE= k8s.io/kube-openapi v0.0.0-20250910181357-589584f1c912 h1:Y3gxNAuB0OBLImH611+UDZcmKS3g6CthxToOb37KgwE= diff --git a/pkg/controller/builder/builder.go b/pkg/controller/builder/builder.go index f5827d7..b4edada 100644 --- a/pkg/controller/builder/builder.go +++ b/pkg/controller/builder/builder.go @@ -96,6 +96,7 @@ func Build(ctx context.Context, conf Configuration, optsFuncs ...FuncOption) (*c Recorder: event.NewAPIRecorder(rec), ThrottledRecorder: event.NewAPIRecorder(throttleRec), Logger: opts.logger, + Metrics: opts.metricsRecorder, ListWatcher: opts.listWatcher, GlobalRateLimiter: opts.globalRateLimiter, MetricsServer: metricsServer, diff --git a/pkg/controller/builder/options.go b/pkg/controller/builder/options.go index b36cb99..ebb9c3e 100644 --- a/pkg/controller/builder/options.go +++ b/pkg/controller/builder/options.go @@ -10,6 +10,7 @@ import ( "github.com/krateoplatformops/unstructured-runtime/pkg/meta" metricsserver "github.com/krateoplatformops/unstructured-runtime/pkg/metrics/server" "github.com/krateoplatformops/unstructured-runtime/pkg/pluralizer" + "github.com/krateoplatformops/unstructured-runtime/pkg/telemetry" "golang.org/x/time/rate" "k8s.io/client-go/util/workqueue" ) @@ -21,6 +22,7 @@ type options struct { logger logging.Logger listWatcher controller.ListWatcherConfiguration globalRateLimiter workqueue.TypedRateLimiter[any] + metricsRecorder *telemetry.Metrics metrics metricsserver.Options // MaxRetries is the maximum number of retries for a failed reconciliation. // If the value is less than or equal to 0, the default value of 5 will be used. @@ -100,6 +102,12 @@ func WithGlobalRateLimiter(rl workqueue.TypedRateLimiter[any]) func(o *options) } } +func WithTelemetryMetrics(m *telemetry.Metrics) func(o *options) { + return func(o *options) { + o.metricsRecorder = m + } +} + func WithMetrics(m metricsserver.Options) func(o *options) { return func(o *options) { o.metrics = m diff --git a/pkg/controller/controller.go b/pkg/controller/controller.go index ea9eb51..a260197 100644 --- a/pkg/controller/controller.go +++ b/pkg/controller/controller.go @@ -18,6 +18,7 @@ import ( "github.com/krateoplatformops/unstructured-runtime/pkg/logging" "github.com/krateoplatformops/unstructured-runtime/pkg/meta" "github.com/krateoplatformops/unstructured-runtime/pkg/pluralizer" + "github.com/krateoplatformops/unstructured-runtime/pkg/telemetry" "github.com/prometheus/client_golang/prometheus" "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" "k8s.io/apimachinery/pkg/runtime/schema" @@ -73,6 +74,7 @@ type Options struct { Recorder event.Recorder ThrottledRecorder event.Recorder Logger logging.Logger + Metrics *telemetry.Metrics ListWatcher ListWatcherConfiguration Pluralizer pluralizer.PluralizerInterface GlobalRateLimiter workqueue.TypedRateLimiter[any] @@ -121,6 +123,7 @@ type Controller struct { recorder event.Recorder throttledRecorder event.Recorder logger logging.Logger + metrics *telemetry.Metrics externalClient ExternalClient maxRetries int } @@ -156,6 +159,12 @@ func New(sid *shortid.Shortid, opts Options) (*Controller, error) { queue := priorityqueue.New("controller", func(o *priorityqueue.Opts[any]) { o.RateLimiter = opts.GlobalRateLimiter }) + + // Wrap the queue with metrics instrumentation if metrics are available + var finalQueue priorityqueue.PriorityQueue[any] = queue + if opts.Metrics != nil { + finalQueue = NewInstrumentedQueue(queue, opts.Metrics) + } items := &sync.Map{} lw, err := listwatcher.Create(listwatcher.CreateOption{ @@ -482,8 +491,9 @@ func New(sid *shortid.Shortid, opts Options) (*Controller, error) { recorder: opts.Recorder, throttledRecorder: opts.ThrottledRecorder, logger: opts.Logger, + metrics: opts.Metrics, informer: informer, - queue: queue, + queue: finalQueue, pluralizer: opts.Pluralizer, metricsServer: opts.MetricsServer, maxRetries: opts.MaxRetries, diff --git a/pkg/controller/queue_instrumentation.go b/pkg/controller/queue_instrumentation.go new file mode 100644 index 0000000..29db3bb --- /dev/null +++ b/pkg/controller/queue_instrumentation.go @@ -0,0 +1,185 @@ +package controller + +import ( + "context" + "sync" + "time" + + "github.com/krateoplatformops/unstructured-runtime/pkg/controller/priorityqueue" +) + +// QueueMetricsRecorder records queue-related OTEL telemetry. +type QueueMetricsRecorder interface { + AddQueueDepth(ctx context.Context, delta int64) + RecordQueueOldestItemAge(ctx context.Context, d time.Duration) +} + +// InstrumentedQueue wraps a priority queue to record OTEL metrics. +type InstrumentedQueue[T comparable] struct { + queue priorityqueue.PriorityQueue[T] + recorder QueueMetricsRecorder + lock sync.Mutex + enqueuedAt map[T]time.Time + done chan struct{} +} + +// NewInstrumentedQueue creates a queue that records OTEL metrics. +func NewInstrumentedQueue[T comparable]( + queue priorityqueue.PriorityQueue[T], + recorder QueueMetricsRecorder, +) *InstrumentedQueue[T] { + iq := &InstrumentedQueue[T]{ + queue: queue, + recorder: recorder, + enqueuedAt: make(map[T]time.Time), + done: make(chan struct{}), + } + + // Start background routine to periodically record oldest item age + if recorder != nil { + go iq.recordOldestItemAgeLoop() + } + + return iq +} + +// recordOldestItemAgeLoop periodically records the age of the oldest item in the queue. +func (iq *InstrumentedQueue[T]) recordOldestItemAgeLoop() { + ticker := time.NewTicker(500 * time.Millisecond) + defer ticker.Stop() + + for { + select { + case <-iq.done: + return + case <-ticker.C: + iq.recordOldestItemAge() + } + } +} + +// recordOldestItemAge records the age of the oldest queued item. +func (iq *InstrumentedQueue[T]) recordOldestItemAge() { + iq.lock.Lock() + defer iq.lock.Unlock() + + var oldest time.Duration + now := time.Now() + for _, enqueuedTime := range iq.enqueuedAt { + age := now.Sub(enqueuedTime) + if age > oldest { + oldest = age + } + } + + if oldest > 0 && iq.recorder != nil { + iq.recorder.RecordQueueOldestItemAge(context.Background(), oldest) + } +} + +// Add adds an item to the queue and records queue depth. +func (iq *InstrumentedQueue[T]) Add(item T) { + iq.recordEnqueueAt(item) + iq.queue.Add(item) +} + +// AddWithOpts adds items with options and records queue depth. +func (iq *InstrumentedQueue[T]) AddWithOpts(o priorityqueue.AddOpts, items ...T) { + for _, item := range items { + iq.recordEnqueueAt(item) + } + iq.queue.AddWithOpts(o, items...) +} + +// AddAfter adds an item after a duration. +func (iq *InstrumentedQueue[T]) AddAfter(item T, duration time.Duration) { + iq.recordEnqueueAt(item) + iq.queue.AddAfter(item, duration) +} + +// AddRateLimited adds an item with rate limiting. +func (iq *InstrumentedQueue[T]) AddRateLimited(item T) { + iq.recordEnqueueAt(item) + iq.queue.AddRateLimited(item) +} + +// Get retrieves an item from the queue. +func (iq *InstrumentedQueue[T]) Get() (item T, shutdown bool) { + item, shutdown = iq.queue.Get() + if !shutdown { + iq.recordDequeueAt(item) + } + return item, shutdown +} + +// GetWithPriority retrieves an item with its priority. +func (iq *InstrumentedQueue[T]) GetWithPriority() (item T, priority int, shutdown bool) { + item, priority, shutdown = iq.queue.GetWithPriority() + if !shutdown { + iq.recordDequeueAt(item) + } + return item, priority, shutdown +} + +// Done marks an item as done. +func (iq *InstrumentedQueue[T]) Done(item T) { + iq.queue.Done(item) +} + +// ShutDown shuts down the queue. +func (iq *InstrumentedQueue[T]) ShutDown() { + close(iq.done) + iq.queue.ShutDown() +} + +// ShutDownWithDrain shuts down the queue after draining it. +func (iq *InstrumentedQueue[T]) ShutDownWithDrain() { + close(iq.done) + iq.queue.ShutDownWithDrain() +} + +// ShuttingDown returns true if the queue is shutting down. +func (iq *InstrumentedQueue[T]) ShuttingDown() bool { + return iq.queue.ShuttingDown() +} + +// Len returns the number of items in the queue. +func (iq *InstrumentedQueue[T]) Len() int { + return iq.queue.Len() +} + +// NumRequeues returns the number of requeues for an item. +func (iq *InstrumentedQueue[T]) NumRequeues(item T) int { + return iq.queue.NumRequeues(item) +} + +// Forget removes an item from retry tracking. +func (iq *InstrumentedQueue[T]) Forget(item T) { + iq.queue.Forget(item) +} + +// recordEnqueueAt records when an item is enqueued. +func (iq *InstrumentedQueue[T]) recordEnqueueAt(item T) { + iq.lock.Lock() + defer iq.lock.Unlock() + + if _, exists := iq.enqueuedAt[item]; !exists { + iq.enqueuedAt[item] = time.Now() + if iq.recorder != nil { + iq.recorder.AddQueueDepth(context.Background(), 1) + } + } +} + +// recordDequeueAt records when an item is dequeued. +func (iq *InstrumentedQueue[T]) recordDequeueAt(item T) { + iq.lock.Lock() + defer iq.lock.Unlock() + + if _, exists := iq.enqueuedAt[item]; exists { + delete(iq.enqueuedAt, item) + if iq.recorder != nil { + iq.recorder.AddQueueDepth(context.Background(), -1) + } + } +} diff --git a/pkg/controller/worker.go b/pkg/controller/worker.go index ece126d..e349876 100644 --- a/pkg/controller/worker.go +++ b/pkg/controller/worker.go @@ -112,9 +112,16 @@ func (c *Controller) runWorker(ctx context.Context) { c.queue.Forget(obj) c.queue.Done(obj) runtimeUtil.HandleError(fmt.Errorf("unexpected object in queue: %v", obj)) + if c.metrics != nil { + c.metrics.AddReconcileInFlight(-1) + } continue } + if c.metrics != nil { + c.metrics.AddReconcileInFlight(1) + } + dig := ctrlevent.DigestForEvent(ev) // Start timing for processing @@ -125,11 +132,17 @@ func (c *Controller) runWorker(ctx context.Context) { var queueWaitDuration time.Duration if !ev.QueuedAt.IsZero() { queueWaitDuration = processingStartTime.Sub(ev.QueuedAt) + if c.metrics != nil { + c.metrics.RecordQueueWaitDuration(ctx, queueWaitDuration) + } } traceId, err := shortid.Generate() if err != nil { runtimeUtil.HandleError(fmt.Errorf("cannot generate trace ID: %v", err)) + if c.metrics != nil { + c.metrics.AddReconcileInFlight(-1) + } continue } ctx = contextutils.BuildContext(ctx, contextutils.WithLogger(c.logger), contextutils.WithTraceId(traceId)) @@ -145,6 +158,16 @@ func (c *Controller) runWorker(ctx context.Context) { } else { totalDuration = processingDuration } + if c.metrics != nil { + c.metrics.RecordQueueWorkDuration(ctx, processingDuration) + c.metrics.RecordReconcileDuration(ctx, totalDuration) + if err == nil { + c.metrics.IncReconcileSuccess(ctx) + } else { + c.metrics.IncReconcileFailure(ctx) + c.metrics.IncReconcileErrorRequeue(ctx) + } + } // Enhanced debug logging c.logger.WithValues( @@ -169,6 +192,9 @@ func (c *Controller) runWorker(ctx context.Context) { c.queue.Forget(ev) c.queue.Done(ev) c.handleErr(err, ev, priority) + if c.metrics != nil { + c.metrics.AddReconcileInFlight(-1) + } } } @@ -359,7 +385,14 @@ func (c *Controller) handleObserve(ctx context.Context, ref objectref.ObjectRef) return err } + start := time.Now() observation, actionErr := c.externalClient.Observe(ctx, el) + if c.metrics != nil { + c.metrics.RecordObserveDuration(ctx, time.Since(start)) + if actionErr != nil { + c.metrics.IncObserveFailure(ctx) + } + } if actionErr != nil { c.recordEvent(el, event.Warning(reasonCannotObserve, actionObserveExternalResource, actionErr)) log.Error(actionErr, "Cannot observe external resource") @@ -506,7 +539,14 @@ func (c *Controller) handleCreate(ctx context.Context, ref objectref.ObjectRef) return err } + start := time.Now() actionErr := c.externalClient.Create(ctx, el) + if c.metrics != nil { + c.metrics.RecordCreateDuration(ctx, time.Since(start)) + if actionErr != nil { + c.metrics.IncCreateFailure(ctx) + } + } if actionErr != nil { c.recordEvent(el, event.Warning(reasonCannotCreate, actionCreateExternalResource, actionErr)) log.Error(actionErr, "Cannot create external resource") @@ -622,7 +662,14 @@ func (c *Controller) handleUpdate(ctx context.Context, ref objectref.ObjectRef) return err } + start := time.Now() actionErr := c.externalClient.Update(ctx, el) + if c.metrics != nil { + c.metrics.RecordUpdateDuration(ctx, time.Since(start)) + if actionErr != nil { + c.metrics.IncUpdateFailure(ctx) + } + } if actionErr != nil { c.recordEvent(el, event.Warning(reasonCannotUpdate, actionUpdateEvent, actionErr)) log.Error(actionErr, "Cannot update external resource") @@ -690,7 +737,14 @@ func (c *Controller) handleDelete(ctx context.Context, ref objectref.ObjectRef) log.Error(err, "Cannot fetch managed resource") return err } + start := time.Now() actionErr := c.externalClient.Delete(ctx, el) + if c.metrics != nil { + c.metrics.RecordDeleteDuration(ctx, time.Since(start)) + if actionErr != nil { + c.metrics.IncDeleteFailure(ctx) + } + } if actionErr != nil { log.Error(actionErr, "Cannot delete external resource") c.recordEvent(el, event.Warning(reasonCannotDelete, actionDeleteExternalResource, actionErr)) @@ -751,11 +805,21 @@ func (c *Controller) handleDelete(ctx context.Context, ref objectref.ObjectRef) return nil } -func (c *Controller) fetch(ctx context.Context, ref objectref.ObjectRef, clean bool) (*unstructured.Unstructured, error) { +func (c *Controller) fetch(ctx context.Context, ref objectref.ObjectRef, clean bool) (res *unstructured.Unstructured, err error) { if c.dynamicClient == nil { return nil, fmt.Errorf("dynamic client is not configured") } - res, err := c.dynamicClient.Resource(c.gvr). + start := time.Now() + defer func() { + if c.metrics == nil { + return + } + c.metrics.RecordGetDuration(ctx, time.Since(start)) + if err != nil { + c.metrics.IncGetFailure(ctx) + } + }() + res, err = c.dynamicClient.Resource(c.gvr). Namespace(ref.Namespace). Get(ctx, ref.Name, metav1.GetOptions{}) if err == nil { diff --git a/pkg/telemetry/metrics.go b/pkg/telemetry/metrics.go new file mode 100644 index 0000000..73fba89 --- /dev/null +++ b/pkg/telemetry/metrics.go @@ -0,0 +1,462 @@ +package telemetry + +import ( + "context" + "sync/atomic" + "time" + + "github.com/krateoplatformops/unstructured-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 = "unstructured-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 runtime metrics for +// unstructured-runtime consumers. +type Metrics struct { + startupSuccess metric.Int64Counter + startupFailure metric.Int64Counter + + getDuration metric.Float64Histogram + getFailure metric.Int64Counter + + connectDuration metric.Float64Histogram + connectFailure metric.Int64Counter + + reconcileDuration metric.Float64Histogram + queueDepth metric.Int64UpDownCounter + queueWaitDuration metric.Float64Histogram + queueOldestItemAge metric.Float64Histogram + queueWorkDuration metric.Float64Histogram + reconcileInFlight metric.Int64ObservableGauge + reconcileInFlightCount atomic.Int64 + + reconcileSuccess metric.Int64Counter + reconcileFailure metric.Int64Counter + reconcileRequeueAfter metric.Int64Counter + reconcileRequeueImmediate metric.Int64Counter + reconcileErrorRequeue metric.Int64Counter + queueRequeueTotal 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/unstructured-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("unstructured_runtime.startup.success"); err != nil { + return nil, err + } + if m.startupFailure, err = meter.Int64Counter("unstructured_runtime.startup.failure"); err != nil { + return nil, err + } + if m.getDuration, err = meter.Float64Histogram("unstructured_runtime.reconcile.get.duration_seconds"); err != nil { + return nil, err + } + if m.getFailure, err = meter.Int64Counter("unstructured_runtime.reconcile.get.failure"); err != nil { + return nil, err + } + if m.reconcileDuration, err = meter.Float64Histogram("unstructured_runtime.reconcile.duration_seconds"); err != nil { + return nil, err + } + if m.queueDepth, err = meter.Int64UpDownCounter("unstructured_runtime.reconcile.queue.depth"); err != nil { + return nil, err + } + if m.queueWaitDuration, err = meter.Float64Histogram("unstructured_runtime.reconcile.queue.wait.duration_seconds"); err != nil { + return nil, err + } + if m.queueOldestItemAge, err = meter.Float64Histogram("unstructured_runtime.reconcile.queue.oldest_item_age_seconds"); err != nil { + return nil, err + } + if m.queueWorkDuration, err = meter.Float64Histogram("unstructured_runtime.reconcile.queue.work.duration_seconds"); err != nil { + return nil, err + } + if m.reconcileInFlight, err = meter.Int64ObservableGauge("unstructured_runtime.reconcile.in_flight"); err != nil { + return nil, err + } + if m.reconcileSuccess, err = meter.Int64Counter("unstructured_runtime.reconcile.success"); err != nil { + return nil, err + } + if m.reconcileFailure, err = meter.Int64Counter("unstructured_runtime.reconcile.failure"); err != nil { + return nil, err + } + if m.reconcileRequeueAfter, err = meter.Int64Counter("unstructured_runtime.reconcile.requeue.after"); err != nil { + return nil, err + } + if m.reconcileRequeueImmediate, err = meter.Int64Counter("unstructured_runtime.reconcile.requeue.immediate"); err != nil { + return nil, err + } + if m.reconcileErrorRequeue, err = meter.Int64Counter("unstructured_runtime.reconcile.requeue.error"); err != nil { + return nil, err + } + if m.queueRequeueTotal, err = meter.Int64Counter("unstructured_runtime.reconcile.queue.requeues"); err != nil { + return nil, err + } + if m.connectDuration, err = meter.Float64Histogram("unstructured_runtime.external.connect.duration_seconds"); err != nil { + return nil, err + } + if m.connectFailure, err = meter.Int64Counter("unstructured_runtime.external.connect.failure"); err != nil { + return nil, err + } + if m.observeDuration, err = meter.Float64Histogram("unstructured_runtime.external.observe.duration_seconds"); err != nil { + return nil, err + } + if m.observeFailure, err = meter.Int64Counter("unstructured_runtime.external.observe.failure"); err != nil { + return nil, err + } + if m.createDuration, err = meter.Float64Histogram("unstructured_runtime.external.create.duration_seconds"); err != nil { + return nil, err + } + if m.createFailure, err = meter.Int64Counter("unstructured_runtime.external.create.failure"); err != nil { + return nil, err + } + if m.updateDuration, err = meter.Float64Histogram("unstructured_runtime.external.update.duration_seconds"); err != nil { + return nil, err + } + if m.updateFailure, err = meter.Int64Counter("unstructured_runtime.external.update.failure"); err != nil { + return nil, err + } + if m.deleteDuration, err = meter.Float64Histogram("unstructured_runtime.external.delete.duration_seconds"); err != nil { + return nil, err + } + if m.deleteFailure, err = meter.Int64Counter("unstructured_runtime.external.delete.failure"); err != nil { + return nil, err + } + if m.finalizerAddDuration, err = meter.Float64Histogram("unstructured_runtime.finalizer.add.duration_seconds"); err != nil { + return nil, err + } + if m.finalizerAddFailure, err = meter.Int64Counter("unstructured_runtime.finalizer.add.failure"); err != nil { + return nil, err + } + if m.finalizerRemoveDuration, err = meter.Float64Histogram("unstructured_runtime.finalizer.remove.duration_seconds"); err != nil { + return nil, err + } + if m.finalizerRemoveFailure, err = meter.Int64Counter("unstructured_runtime.finalizer.remove.failure"); err != nil { + return nil, err + } + if m.statusUpdateDuration, err = meter.Float64Histogram("unstructured_runtime.status.update.duration_seconds"); err != nil { + return nil, err + } + if m.statusUpdateFailure, err = meter.Int64Counter("unstructured_runtime.status.update.failure"); err != nil { + return nil, err + } + if m.connectDuration, err = meter.Float64Histogram("unstructured_runtime.external.connect.duration_seconds"); err != nil { + return nil, err + } + if m.connectFailure, err = meter.Int64Counter("unstructured_runtime.external.connect.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) 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) 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) +} + +func (m *Metrics) AddQueueDepth(ctx context.Context, delta int64) { + if m == nil { + 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) 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) +} diff --git a/pkg/telemetry/metrics_test.go b/pkg/telemetry/metrics_test.go new file mode 100644 index 0000000..7fce20b --- /dev/null +++ b/pkg/telemetry/metrics_test.go @@ -0,0 +1,130 @@ +package telemetry + +import ( + "context" + "testing" + "time" + + "github.com/krateoplatformops/unstructured-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/unstructured-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.RecordConnectDuration(ctx, 20*time.Millisecond) + metrics.IncConnectFailure(ctx) + metrics.RecordObserveDuration(ctx, 10*time.Millisecond) + metrics.IncObserveFailure(ctx) + metrics.RecordCreateDuration(ctx, 11*time.Millisecond) + metrics.IncCreateFailure(ctx) + metrics.RecordUpdateDuration(ctx, 12*time.Millisecond) + metrics.IncUpdateFailure(ctx) + metrics.RecordDeleteDuration(ctx, 13*time.Millisecond) + metrics.IncDeleteFailure(ctx) + metrics.RecordFinalizerAddDuration(ctx, 14*time.Millisecond) + metrics.IncFinalizerAddFailure(ctx) + metrics.RecordFinalizerRemoveDuration(ctx, 15*time.Millisecond) + metrics.IncFinalizerRemoveFailure(ctx) + metrics.RecordStatusUpdateDuration(ctx, 16*time.Millisecond) + metrics.IncStatusUpdateFailure(ctx) + metrics.IncReconcileSuccess(ctx) + metrics.IncReconcileFailure(ctx) + metrics.IncReconcileRequeueAfter(ctx) + metrics.IncReconcileRequeueImmediate(ctx) + metrics.IncReconcileErrorRequeue(ctx) + metrics.AddReconcileInFlight(1) + + var rm metricdata.ResourceMetrics + if err := reader.Collect(ctx, &rm); err != nil { + t.Fatalf("reader.Collect() returned error: %v", err) + } + + mustHaveMetric(t, rm, "unstructured_runtime.startup.success") + mustHaveMetric(t, rm, "unstructured_runtime.reconcile.get.duration_seconds") + mustHaveMetric(t, rm, "unstructured_runtime.reconcile.get.failure") + mustHaveMetric(t, rm, "unstructured_runtime.reconcile.duration_seconds") + mustHaveMetric(t, rm, "unstructured_runtime.reconcile.queue.depth") + mustHaveMetric(t, rm, "unstructured_runtime.reconcile.queue.wait.duration_seconds") + mustHaveMetric(t, rm, "unstructured_runtime.reconcile.queue.oldest_item_age_seconds") + mustHaveMetric(t, rm, "unstructured_runtime.reconcile.queue.work.duration_seconds") + mustHaveMetric(t, rm, "unstructured_runtime.reconcile.queue.requeues") + mustHaveMetric(t, rm, "unstructured_runtime.reconcile.success") + mustHaveMetric(t, rm, "unstructured_runtime.reconcile.failure") + mustHaveMetric(t, rm, "unstructured_runtime.reconcile.requeue.after") + mustHaveMetric(t, rm, "unstructured_runtime.reconcile.requeue.immediate") + mustHaveMetric(t, rm, "unstructured_runtime.reconcile.requeue.error") + mustHaveMetric(t, rm, "unstructured_runtime.reconcile.in_flight") + mustHaveMetric(t, rm, "unstructured_runtime.external.connect.duration_seconds") + mustHaveMetric(t, rm, "unstructured_runtime.external.connect.failure") + mustHaveMetric(t, rm, "unstructured_runtime.external.observe.duration_seconds") + mustHaveMetric(t, rm, "unstructured_runtime.external.observe.failure") + mustHaveMetric(t, rm, "unstructured_runtime.external.create.duration_seconds") + mustHaveMetric(t, rm, "unstructured_runtime.external.create.failure") + mustHaveMetric(t, rm, "unstructured_runtime.external.update.duration_seconds") + mustHaveMetric(t, rm, "unstructured_runtime.external.update.failure") + mustHaveMetric(t, rm, "unstructured_runtime.external.delete.duration_seconds") + mustHaveMetric(t, rm, "unstructured_runtime.external.delete.failure") + mustHaveMetric(t, rm, "unstructured_runtime.finalizer.add.duration_seconds") + mustHaveMetric(t, rm, "unstructured_runtime.finalizer.add.failure") + mustHaveMetric(t, rm, "unstructured_runtime.finalizer.remove.duration_seconds") + mustHaveMetric(t, rm, "unstructured_runtime.finalizer.remove.failure") + mustHaveMetric(t, rm, "unstructured_runtime.status.update.duration_seconds") + mustHaveMetric(t, rm, "unstructured_runtime.status.update.failure") +} + +func mustHaveMetric(t *testing.T, rm metricdata.ResourceMetrics, name string) { + t.Helper() + if !hasMetric(rm, name) { + t.Fatalf("expected %s metric to be collected", name) + } +} + +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 db892da890d9ba183c68331503cb441f23120892 Mon Sep 17 00:00:00 2001 From: Matteo Gastaldello Date: Thu, 16 Apr 2026 17:33:28 +0200 Subject: [PATCH 2/4] feat: add deployment name to metrics configuration for stable resource identification --- pkg/telemetry/metrics.go | 26 +++++++++++++++++++++----- pkg/telemetry/metrics_test.go | 2 +- 2 files changed, 22 insertions(+), 6 deletions(-) diff --git a/pkg/telemetry/metrics.go b/pkg/telemetry/metrics.go index 73fba89..384a8d7 100644 --- a/pkg/telemetry/metrics.go +++ b/pkg/telemetry/metrics.go @@ -24,11 +24,14 @@ type Config struct { Enabled bool ServiceName string ExportInterval time.Duration + DeploymentName string // Deployment name for stable resource identification } // Metrics exposes a small set of low-cardinality runtime metrics for // unstructured-runtime consumers. type Metrics struct { + log logging.Logger + startupSuccess metric.Int64Counter startupFailure metric.Int64Counter @@ -102,8 +105,19 @@ func Setup(ctx context.Context, log logging.Logger, cfg Config) (*Metrics, func( return nil, nil, err } + // Build resource attributes including deployment name for stable multi-instance identification + attrs := []attribute.KeyValue{ + attribute.String("service.name", serviceName), + } + if cfg.DeploymentName != "" { + attrs = append(attrs, + attribute.String("k8s.deployment.name", cfg.DeploymentName), + attribute.String("service.instance.id", cfg.DeploymentName), + ) + } + res, err := resource.Merge(resource.Default(), - resource.NewSchemaless(attribute.String("service.name", serviceName))) + resource.NewSchemaless(attrs...)) if err != nil { return nil, nil, err } @@ -115,21 +129,23 @@ func Setup(ctx context.Context, log logging.Logger, cfg Config) (*Metrics, func( ) meter := provider.Meter("github.com/krateoplatformops/unstructured-runtime") - metrics, err := newMetrics(meter) + metrics, err := newMetrics(meter, log) if err != nil { _ = provider.Shutdown(ctx) return nil, nil, err } otel.SetMeterProvider(provider) - log.Info("OpenTelemetry metrics initialized") + log.Info("OpenTelemetry metrics initialized", "deploymentName", cfg.DeploymentName, "serviceName", serviceName, "exportInterval", exportInterval) return metrics, provider.Shutdown, nil } -func newMetrics(meter metric.Meter) (*Metrics, error) { +func newMetrics(meter metric.Meter, log logging.Logger) (*Metrics, error) { var err error - m := &Metrics{} + m := &Metrics{ + log: log, + } if m.startupSuccess, err = meter.Int64Counter("unstructured_runtime.startup.success"); err != nil { return nil, err diff --git a/pkg/telemetry/metrics_test.go b/pkg/telemetry/metrics_test.go index 7fce20b..910618a 100644 --- a/pkg/telemetry/metrics_test.go +++ b/pkg/telemetry/metrics_test.go @@ -36,7 +36,7 @@ func TestNewMetricsRecordsData(t *testing.T) { } }) - metrics, err := newMetrics(provider.Meter("github.com/krateoplatformops/unstructured-runtime/test")) + metrics, err := newMetrics(provider.Meter("github.com/krateoplatformops/unstructured-runtime/test"), logging.NewNopLogger()) if err != nil { t.Fatalf("newMetrics() returned error: %v", err) } From 88d75b89595a99a6d0724c6566d6cc99ec5a25c3 Mon Sep 17 00:00:00 2001 From: Matteo Gastaldello Date: Fri, 17 Apr 2026 18:20:29 +0200 Subject: [PATCH 3/4] feat: add deployment name to metrics for improved instance tracking --- go.mod | 4 ++-- go.sum | 8 ++++---- pkg/telemetry/metrics.go | 27 +++++++++++++++++---------- 3 files changed, 23 insertions(+), 16 deletions(-) diff --git a/go.mod b/go.mod index b878c03..536ff71 100644 --- a/go.mod +++ b/go.mod @@ -9,7 +9,7 @@ require ( github.com/google/btree v1.1.3 github.com/google/go-cmp v0.7.0 github.com/google/gofuzz v1.2.0 - github.com/krateoplatformops/plumbing v1.7.2 + github.com/krateoplatformops/plumbing v1.7.3 github.com/onsi/ginkgo/v2 v2.27.2 github.com/onsi/gomega v1.38.2 github.com/prometheus/client_golang v1.23.2 @@ -56,7 +56,7 @@ require ( github.com/json-iterator/go v1.1.12 // indirect github.com/kylelemons/godebug v1.1.0 // indirect github.com/mailru/easyjson v0.9.0 // indirect - github.com/moby/spdystream v0.5.0 // indirect + github.com/moby/spdystream v0.5.1 // indirect github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect github.com/modern-go/reflect2 v1.0.3-0.20250322232337-35a7c28c31ee // indirect github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect diff --git a/go.sum b/go.sum index 9a8adb6..07c1ef3 100644 --- a/go.sum +++ b/go.sum @@ -80,8 +80,8 @@ github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE= github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk= github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY= github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE= -github.com/krateoplatformops/plumbing v1.7.2 h1:I9rvpKPQ1eswRZkdwCT+xfdDRLODSsL1c7a6g8K3S9c= -github.com/krateoplatformops/plumbing v1.7.2/go.mod h1:UrYvAvnfacOiOLlQbIg6J08zYC+XJ9ZMKbecQq+4jV4= +github.com/krateoplatformops/plumbing v1.7.3 h1:rQ8j5r47j4i1SETvIQgnlJ29357g86QjAhpKsGxMlfc= +github.com/krateoplatformops/plumbing v1.7.3/go.mod h1:TE35fxU8TWaTrR0Jw+b+m48x+zkBRxyvP1uC7ZoghOQ= github.com/kylelemons/godebug v1.1.0 h1:RPNrshWIDI6G2gRW9EHilWtl7Z6Sb1BR0xunSBf0SNc= github.com/kylelemons/godebug v1.1.0/go.mod h1:9/0rRGxNHcop5bhtWyNeEfOS8JIWk580+fNqagV/RAw= github.com/mailru/easyjson v0.9.0 h1:PrnmzHw7262yW8sTBwxi1PdJA3Iw/EKBa8psRf7d9a4= @@ -90,8 +90,8 @@ github.com/maruel/natural v1.1.1 h1:Hja7XhhmvEFhcByqDoHz9QZbkWey+COd9xWfCfn1ioo= github.com/maruel/natural v1.1.1/go.mod h1:v+Rfd79xlw1AgVBjbO0BEQmptqb5HvL/k9GRHB7ZKEg= github.com/mfridman/tparse v0.18.0 h1:wh6dzOKaIwkUGyKgOntDW4liXSo37qg5AXbIhkMV3vE= github.com/mfridman/tparse v0.18.0/go.mod h1:gEvqZTuCgEhPbYk/2lS3Kcxg1GmTxxU7kTC8DvP0i/A= -github.com/moby/spdystream v0.5.0 h1:7r0J1Si3QO/kjRitvSLVVFUjxMEb/YLj6S9FF62JBCU= -github.com/moby/spdystream v0.5.0/go.mod h1:xBAYlnt/ay+11ShkdFKNAG7LsyK/tmNBVvVOwrfMgdI= +github.com/moby/spdystream v0.5.1 h1:9sNYeYZUcci9R6/w7KDaFWEWeV4LStVG78Mpyq/Zm/Y= +github.com/moby/spdystream v0.5.1/go.mod h1:xBAYlnt/ay+11ShkdFKNAG7LsyK/tmNBVvVOwrfMgdI= github.com/modern-go/concurrent v0.0.0-20180228061459-e0a39a4cb421/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q= github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd h1:TRLaZ9cD/w8PVh93nsPXa1VrQ6jlwL5oN8l14QlcNfg= github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q= diff --git a/pkg/telemetry/metrics.go b/pkg/telemetry/metrics.go index 384a8d7..02e60f5 100644 --- a/pkg/telemetry/metrics.go +++ b/pkg/telemetry/metrics.go @@ -30,7 +30,8 @@ type Config struct { // Metrics exposes a small set of low-cardinality runtime metrics for // unstructured-runtime consumers. type Metrics struct { - log logging.Logger + log logging.Logger + deploymentName string // For tracking instance identity in exported metrics startupSuccess metric.Int64Counter startupFailure metric.Int64Counter @@ -129,7 +130,7 @@ func Setup(ctx context.Context, log logging.Logger, cfg Config) (*Metrics, func( ) meter := provider.Meter("github.com/krateoplatformops/unstructured-runtime") - metrics, err := newMetrics(meter, log) + metrics, err := newMetrics(meter, log, cfg.DeploymentName) if err != nil { _ = provider.Shutdown(ctx) return nil, nil, err @@ -141,10 +142,11 @@ func Setup(ctx context.Context, log logging.Logger, cfg Config) (*Metrics, func( return metrics, provider.Shutdown, nil } -func newMetrics(meter metric.Meter, log logging.Logger) (*Metrics, error) { +func newMetrics(meter metric.Meter, log logging.Logger, deploymentName string) (*Metrics, error) { var err error m := &Metrics{ - log: log, + log: log, + deploymentName: deploymentName, } if m.startupSuccess, err = meter.Int64Counter("unstructured_runtime.startup.success"); err != nil { @@ -243,12 +245,6 @@ func newMetrics(meter metric.Meter, log logging.Logger) (*Metrics, error) { if m.statusUpdateFailure, err = meter.Int64Counter("unstructured_runtime.status.update.failure"); err != nil { return nil, err } - if m.connectDuration, err = meter.Float64Histogram("unstructured_runtime.external.connect.duration_seconds"); err != nil { - return nil, err - } - if m.connectFailure, err = meter.Int64Counter("unstructured_runtime.external.connect.failure"); err != nil { - return nil, err - } _, err = meter.RegisterCallback(func(_ context.Context, observer metric.Observer) error { observer.ObserveInt64(m.reconcileInFlight, m.reconcileInFlightCount.Load()) @@ -261,6 +257,17 @@ func newMetrics(meter metric.Meter, log logging.Logger) (*Metrics, error) { return m, nil } +// instanceAttrs returns metric attributes for instance identification. +// This enables metrics exported to Prometheus to include service.instance.id as exported_instance. +func (m *Metrics) instanceAttrs() []attribute.KeyValue { + if m == nil || m.deploymentName == "" { + return nil + } + return []attribute.KeyValue{ + attribute.String("service.instance.id", m.deploymentName), + } +} + func (m *Metrics) IncStartupSuccess(ctx context.Context) { if m == nil { return From d6a1f3db9b7e8c574993659ea2ff331a9795a4d7 Mon Sep 17 00:00:00 2001 From: Matteo Gastaldello Date: Fri, 17 Apr 2026 18:26:57 +0200 Subject: [PATCH 4/4] feat: include deployment name in newMetrics function for enhanced tracking --- pkg/telemetry/metrics_test.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pkg/telemetry/metrics_test.go b/pkg/telemetry/metrics_test.go index 910618a..a2a4a41 100644 --- a/pkg/telemetry/metrics_test.go +++ b/pkg/telemetry/metrics_test.go @@ -36,7 +36,7 @@ func TestNewMetricsRecordsData(t *testing.T) { } }) - metrics, err := newMetrics(provider.Meter("github.com/krateoplatformops/unstructured-runtime/test"), logging.NewNopLogger()) + metrics, err := newMetrics(provider.Meter("github.com/krateoplatformops/unstructured-runtime/test"), logging.NewNopLogger(), "test-deployment") if err != nil { t.Fatalf("newMetrics() returned error: %v", err) }