From 85f8707479ca6c5496947b9311ff74201d189c01 Mon Sep 17 00:00:00 2001 From: Fredrik Steen Date: Wed, 10 Jun 2026 14:55:36 +0200 Subject: [PATCH] feat: debounced rendering and improved robustness - Implement debounced render loop to coalesce burst events into single render - Add command execution timeout with context - Atomic file writes with temp-file-and-rename pattern - Kubeconfig resolution with in-cluster fallback - Improve hash calculation with proper field separators - Validate configuration numeric fields (resyncInterval, minNodeCount, commandTimeout) - Handle node external IP loss gracefully with safety net --- .gitignore | 3 + Dockerfile | 4 +- go.mod | 20 +- go.sum | 48 +++-- main.go | 353 +++++++++++++++++++++------------ main_test.go | 537 ++++++++++++++++++++++++++++++++++++++++++++++++++- 6 files changed, 815 insertions(+), 150 deletions(-) diff --git a/.gitignore b/.gitignore index 46464d6..f1d7813 100644 --- a/.gitignore +++ b/.gitignore @@ -11,6 +11,9 @@ k8s-node-external-ip-watcher # Output of the go coverage tool *.out +# Local cluster credentials - never commit +k3d/kubeconfig + # Config files (keep examples) config.yaml *.tmpl diff --git a/Dockerfile b/Dockerfile index 2f8a23b..89cbe8d 100644 --- a/Dockerfile +++ b/Dockerfile @@ -1,11 +1,11 @@ FROM golang:1 AS builder WORKDIR /build +COPY go.mod go.sum ./ +RUN go mod download COPY . . RUN set -eux ; \ apt-get update; \ apt-get install -y --no-install-recommends ca-certificates; \ - go mod download; \ - go mod tidy; \ mkdir -p output; \ mkdir -p config; \ CGO_ENABLED=0 GOOS=linux go build -a -installsuffix cgo -ldflags '-extldflags "-static"' -o k8s-node-external-ip-watcher . diff --git a/go.mod b/go.mod index c5cad14..42f005c 100644 --- a/go.mod +++ b/go.mod @@ -3,12 +3,16 @@ module github.com/nordic-cdn/k8s-node-external-ip-watcher go 1.25.3 require ( + github.com/prometheus/client_golang v1.23.2 gopkg.in/yaml.v3 v3.0.1 k8s.io/api v0.34.1 + k8s.io/apimachinery v0.34.1 k8s.io/client-go v0.34.1 ) require ( + github.com/beorn7/perks v1.0.1 // indirect + github.com/cespare/xxhash/v2 v2.3.0 // indirect github.com/davecgh/go-spew v1.1.1 // indirect github.com/emicklei/go-restful/v3 v3.12.2 // indirect github.com/fxamacker/cbor/v2 v2.9.0 // indirect @@ -28,20 +32,22 @@ require ( github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect github.com/pkg/errors v0.9.1 // indirect github.com/pmezard/go-difflib v1.0.0 // indirect + github.com/prometheus/client_model v0.6.2 // indirect + github.com/prometheus/common v0.66.1 // indirect + github.com/prometheus/procfs v0.16.1 // indirect github.com/spf13/pflag v1.0.6 // indirect github.com/x448/float16 v0.8.4 // indirect go.yaml.in/yaml/v2 v2.4.2 // indirect go.yaml.in/yaml/v3 v3.0.4 // indirect - golang.org/x/net v0.38.0 // indirect - golang.org/x/oauth2 v0.27.0 // indirect - golang.org/x/sys v0.31.0 // indirect - golang.org/x/term v0.30.0 // indirect - golang.org/x/text v0.23.0 // indirect + golang.org/x/net v0.43.0 // indirect + golang.org/x/oauth2 v0.30.0 // indirect + golang.org/x/sys v0.35.0 // indirect + golang.org/x/term v0.34.0 // indirect + golang.org/x/text v0.28.0 // indirect golang.org/x/time v0.9.0 // indirect - google.golang.org/protobuf v1.36.5 // indirect + google.golang.org/protobuf v1.36.8 // indirect gopkg.in/evanphx/json-patch.v4 v4.12.0 // indirect gopkg.in/inf.v0 v0.9.1 // indirect - k8s.io/apimachinery v0.34.1 // indirect k8s.io/klog/v2 v2.130.1 // indirect k8s.io/kube-openapi v0.0.0-20250710124328-f3f2b991d03b // indirect k8s.io/utils v0.0.0-20250604170112-4c0f3b243397 // indirect diff --git a/go.sum b/go.sum index 85113c7..7b25e0e 100644 --- a/go.sum +++ b/go.sum @@ -1,3 +1,7 @@ +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/cespare/xxhash/v2 v2.3.0 h1:UL815xU9SqsFlibzuggzjXhog7bL6oX9BbNZnL2UFvs= +github.com/cespare/xxhash/v2 v2.3.0/go.mod h1:VGX0DQ3Q6kWi7AoAeZDth3/j3BFtOZR5XLFGgcrjCOs= github.com/creack/pty v1.1.9/go.mod h1:oKZEueFk5CKHvIhNR5MUki03XCEU+Q6VDXinZuGJ33E= github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38= github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c= @@ -35,6 +39,8 @@ github.com/json-iterator/go v1.1.12 h1:PV8peI4a0ysnczrg+LtxykD8LfKY9ML6u2jnxaEnr github.com/json-iterator/go v1.1.12/go.mod h1:e30LSqwooZae/UwlEbR2852Gd8hjQvJoHmT4TnhNGBo= github.com/kisielk/errcheck v1.5.0/go.mod h1:pFxgyoBC7bSaBwPgfKdkLd5X25qrDl4LWUI2bnpBCr8= github.com/kisielk/gotool v1.0.0/go.mod h1:XhKaO+MFFWcvkIS/tQcRk01m1F5IRFswLeQ+oQHNcck= +github.com/klauspost/compress v1.18.0 h1:c/Cqfb0r+Yi+JtIEq73FWXVkRonBlf0CRNYc8Zttxdo= +github.com/klauspost/compress v1.18.0/go.mod h1:2Pp+KzxcywXVXMr50+X0Q/Lsb43OQHYWRCY2AiWywWQ= github.com/kr/pretty v0.2.1/go.mod h1:ipq/a2n7PKx3OHsz4KJII5eveXtPO4qwEXGdVfWzfnI= github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE= github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk= @@ -42,6 +48,8 @@ github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ= github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI= github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY= github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE= +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.7.7 h1:UGYAvKxe3sBsEDzO8ZeWOSlIQfWFlxbzLZe7hwFURr0= github.com/mailru/easyjson v0.7.7/go.mod h1:xzfreul335JAWq5oZzymOObrkdz5UnU4kGfJJLY9Nlc= github.com/modern-go/concurrent v0.0.0-20180228061459-e0a39a4cb421/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q= @@ -60,6 +68,14 @@ github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4= github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0= github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4= +github.com/prometheus/client_golang v1.23.2 h1:Je96obch5RDVy3FDMndoUsjAhG5Edi49h0RJWRi/o0o= +github.com/prometheus/client_golang v1.23.2/go.mod h1:Tb1a6LWHB3/SPIzCoaDXI4I8UHKeFTEQ1YCr+0Gyqmg= +github.com/prometheus/client_model v0.6.2 h1:oBsgwpGs7iVziMvrGhE53c/GrLUsZdHnqNwqPLxwZyk= +github.com/prometheus/client_model v0.6.2/go.mod h1:y3m2F6Gdpfy6Ut/GBsUqTWZqCUvMVzSfMLjcu6wAwpE= +github.com/prometheus/common v0.66.1 h1:h5E0h5/Y8niHc5DlaLlWLArTQI7tMrsfQjHV+d9ZoGs= +github.com/prometheus/common v0.66.1/go.mod h1:gcaUsgf3KfRSwHY4dIMXLPV0K/Wg1oZ8+SbZk/HH/dA= +github.com/prometheus/procfs v0.16.1 h1:hZ15bTNuirocR6u0JZ6BAHHmwS1p8B4P6MRqxtzMyRg= +github.com/prometheus/procfs v0.16.1/go.mod h1:teAbpZRB1iIAJYREa1LsoWUXykVXA1KlTmWl8x/U+Is= github.com/rogpeppe/go-internal v1.13.1 h1:KvO1DLK/DRN07sQ1LQKScxyZJuNnedQ5/wKSR38lUII= github.com/rogpeppe/go-internal v1.13.1/go.mod h1:uMEvuHeurkdAXX61udpOXGD/AzZDWNMNyH2VO9fmH0o= github.com/spf13/pflag v1.0.6 h1:jFzHGLGAlb3ruxLB8MhbI6A8+AQX/2eW4qeyNZXNp2o= @@ -73,8 +89,8 @@ github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UV github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg= github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO+kdMU+MU= github.com/stretchr/testify v1.8.1/go.mod h1:w2LPCIKwWwSfY2zedu0+kehJoqGctiVI29o6fzry7u4= -github.com/stretchr/testify v1.10.0 h1:Xv5erBjTwe/5IxqUQTdXv5kgmIvbHo3QQyRwhJsOfJA= -github.com/stretchr/testify v1.10.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY= +github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U= +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= github.com/yuin/goldmark v1.1.27/go.mod h1:3hX8gzYuyVAZsxl0MRgGTJEmQBFcNTphYh9decYSb74= @@ -94,38 +110,38 @@ golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s= golang.org/x/net v0.0.0-20200226121028-0de0cce0169b/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s= golang.org/x/net v0.0.0-20201021035429-f5854403a974/go.mod h1:sp8m0HH+o8qH0wwXwYZr8TS3Oi6o0r6Gce1SSxlDquU= -golang.org/x/net v0.38.0 h1:vRMAPTMaeGqVhG5QyLJHqNDwecKTomGeqbnfZyKlBI8= -golang.org/x/net v0.38.0/go.mod h1:ivrbrMbzFq5J41QOQh0siUuly180yBYtLp+CKbEaFx8= -golang.org/x/oauth2 v0.27.0 h1:da9Vo7/tDv5RH/7nZDz1eMGS/q1Vv1N/7FCrBhI9I3M= -golang.org/x/oauth2 v0.27.0/go.mod h1:onh5ek6nERTohokkhCD/y2cV4Do3fxFHFuAejCkRWT8= +golang.org/x/net v0.43.0 h1:lat02VYK2j4aLzMzecihNvTlJNQUq316m2Mr9rnM6YE= +golang.org/x/net v0.43.0/go.mod h1:vhO1fvI4dGsIjh73sWfUVjj3N7CA9WkKJNQm2svM6Jg= +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.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sync v0.0.0-20190911185100-cd5d95a43a6e/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM= golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY= golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= golang.org/x/sys v0.0.0-20200930185726-fdedc70b468f/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs= -golang.org/x/sys v0.31.0 h1:ioabZlmFYtWhL+TRYpcnNlLwhyxaM9kWTDEmfnprqik= -golang.org/x/sys v0.31.0/go.mod h1:BJP2sWEmIv4KK5OTEluFJCKSidICx8ciO85XgH3Ak8k= -golang.org/x/term v0.30.0 h1:PQ39fJZ+mfadBm0y5WlL4vlM7Sx1Hgf13sMIY2+QS9Y= -golang.org/x/term v0.30.0/go.mod h1:NYYFdzHoI5wRh/h5tDMdMqCqPJZEuNqVR5xJLd/n67g= +golang.org/x/sys v0.35.0 h1:vz1N37gP5bs89s7He8XuIYXpyY0+QlsKmzipCbUtyxI= +golang.org/x/sys v0.35.0/go.mod h1:BJP2sWEmIv4KK5OTEluFJCKSidICx8ciO85XgH3Ak8k= +golang.org/x/term v0.34.0 h1:O/2T7POpk0ZZ7MAzMeWFSg6S5IpWd/RXDlM9hgM3DR4= +golang.org/x/term v0.34.0/go.mod h1:5jC53AEywhIVebHgPVeg0mj8OD3VO9OzclacVrqpaAw= golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ= golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ= -golang.org/x/text v0.23.0 h1:D71I7dUrlY+VX0gQShAThNGHFxZ13dGLBHQLVl1mJlY= -golang.org/x/text v0.23.0/go.mod h1:/BLNzu4aZCJ1+kcD0DNRotWKage4q2rGVAg4o22unh4= +golang.org/x/text v0.28.0 h1:rhazDwis8INMIwQ4tpjLDzUhx6RlXqZNPEM0huQojng= +golang.org/x/text v0.28.0/go.mod h1:U8nCwOR8jO/marOQ0QbDiOngZVEBB7MAiitBuMjXiNU= golang.org/x/time v0.9.0 h1:EsRrnYcQiGH+5FfbgvV4AP7qEZstoyrHB0DzarOQ4ZY= golang.org/x/time v0.9.0/go.mod h1:3BpzKBy/shNhVucY/MWOyx10tF3SFh9QdLuxbVysPQM= golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ= golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo= golang.org/x/tools v0.0.0-20200619180055-7c47624df98f/go.mod h1:EkVYQZoAsY45+roYkvgYkIh4xh/qjgUK9TdY2XT94GE= golang.org/x/tools v0.0.0-20210106214847-113979e3529a/go.mod h1:emZCQorbCU4vsT4fOWvOPXz4eW1wZW4PmDk9uLelYpA= -golang.org/x/tools v0.26.0 h1:v/60pFQmzmT9ExmjDv2gGIfi3OqfKoEP6I5+umXlbnQ= -golang.org/x/tools v0.26.0/go.mod h1:TPVVj70c7JJ3WCazhD8OdXcZg/og+b9+tH/KxylGwH0= +golang.org/x/tools v0.35.0 h1:mBffYraMEf7aa0sB+NuKnuCy8qI/9Bughn8dC2Gu5r0= +golang.org/x/tools v0.35.0/go.mod h1:NKdj5HkL/73byiZSJjqJgKn3ep7KjFkBOkR/Hps3VPw= golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= golang.org/x/xerrors v0.0.0-20200804184101-5ec99f83aff1/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0= -google.golang.org/protobuf v1.36.5 h1:tPhr+woSbjfYvY6/GPufUoYizxw1cF/yFoxJ2fmpwlM= -google.golang.org/protobuf v1.36.5/go.mod h1:9fA7Ob0pmnwhb644+1+CVWFRbNajQ6iRojtC/QF5bRE= +google.golang.org/protobuf v1.36.8 h1:xHScyCOEuuwZEc6UtSOvPbAT4zRh0xcNRYekJwfqyMc= +google.golang.org/protobuf v1.36.8/go.mod h1:fuxRtAxBytpl4zzqUh6/eyUujkJdNiuEkXntxiD/uRU= 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/main.go b/main.go index 3c3b42b..da73e8e 100644 --- a/main.go +++ b/main.go @@ -2,12 +2,14 @@ package main import ( + "bytes" "context" "crypto/sha256" "encoding/hex" "flag" "fmt" "log/slog" + "net" "net/http" "os" "os/exec" @@ -25,6 +27,7 @@ import ( corev1 "k8s.io/api/core/v1" "k8s.io/client-go/informers" "k8s.io/client-go/kubernetes" + "k8s.io/client-go/rest" "k8s.io/client-go/tools/cache" "k8s.io/client-go/tools/clientcmd" ) @@ -61,7 +64,7 @@ var ( currentNodeCount = prometheus.NewGauge( prometheus.GaugeOpts{ Name: "k8s_node_watcher_nodes_current", - Help: "Current number of nodes being monitored", + Help: "Current number of nodes with an external IP", }, ) @@ -93,6 +96,7 @@ type Config struct { ResyncInterval int `yaml:"resyncInterval"` // in seconds MinNodeCount int `yaml:"minNodeCount"` // minimum nodes to prevent empty list MetricsAddr string `yaml:"metricsAddr"` // address for metrics/health HTTP server + CommandTimeout int `yaml:"commandTimeout"` // max command runtime in seconds } // NodeData is the template data @@ -109,6 +113,11 @@ type NodeInfo struct { ExternalIP string } +// defaultDebounce is how long the render loop waits after the first +// pending change so bursts of node events (e.g. the informer replaying +// all nodes at startup) coalesce into a single render+command +const defaultDebounce = 2 * time.Second + // Watcher manages the node watching logic type Watcher struct { config *Config @@ -118,6 +127,8 @@ type Watcher struct { currentHash string nodeIPs map[string]string // node name -> external IP tmpl *template.Template + renderCh chan struct{} // signals the render loop that state changed + debounce time.Duration } func main() { @@ -151,7 +162,11 @@ func main() { ctx, cancel := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM) defer cancel() - httpServer := startHTTPServer(cfg.MetricsAddr, logger) + httpServer, err := startHTTPServer(cfg.MetricsAddr, logger) + if err != nil { + logger.Error("Failed to start HTTP server", "error", err) + os.Exit(1) + } defer func() { shutdownCtx, shutdownCancel := context.WithTimeout(context.Background(), 5*time.Second) defer shutdownCancel() @@ -183,6 +198,7 @@ func loadConfig(configFile, logLevel, kubeConfig, templatePath, outputPath, metr ResyncInterval: 300, // 5 minutes default MinNodeCount: 1, // at least 1 node by default (safety net?) MetricsAddr: "localhost:8089", // default metric listener address + CommandTimeout: 30, // 30 seconds default } // Load from file if it exists @@ -224,6 +240,15 @@ func loadConfig(configFile, logLevel, kubeConfig, templatePath, outputPath, metr if cfg.Command == "" { return nil, fmt.Errorf("command is required") } + if cfg.ResyncInterval < 0 { + return nil, fmt.Errorf("resyncInterval must not be negative, got %d", cfg.ResyncInterval) + } + if cfg.MinNodeCount < 0 { + return nil, fmt.Errorf("minNodeCount must not be negative, got %d", cfg.MinNodeCount) + } + if cfg.CommandTimeout <= 0 { + return nil, fmt.Errorf("commandTimeout must be positive, got %d", cfg.CommandTimeout) + } return cfg, nil } @@ -253,50 +278,78 @@ func setupLogger(level string) *slog.Logger { return slog.New(handler) } -// startHTTPServer starts the HTTP server for metrics and health endpoints -func startHTTPServer(addr string, logger *slog.Logger) *http.Server { +// startHTTPServer starts the HTTP server for metrics and health endpoints. +// The listener is opened synchronously so a bad address fails at startup; +// the returned server's Addr holds the actual listen address +func startHTTPServer(addr string, logger *slog.Logger) (*http.Server, error) { mux := http.NewServeMux() // Simple 200 OK health check endpoint mux.HandleFunc("/healthz", func(w http.ResponseWriter, r *http.Request) { w.WriteHeader(http.StatusOK) - w.Write([]byte("ok\n")) + _, _ = w.Write([]byte("ok\n")) }) // Metrics endpoint mux.Handle("/metrics", promhttp.Handler()) + listener, err := net.Listen("tcp", addr) + if err != nil { + return nil, fmt.Errorf("listen on %s: %w", addr, err) + } + server := &http.Server{ - Addr: addr, - Handler: mux, + Addr: listener.Addr().String(), + Handler: mux, + ReadHeaderTimeout: 5 * time.Second, } go func() { - logger.Info("Starting HTTP server", "addr", addr) - if err := server.ListenAndServe(); err != nil && err != http.ErrServerClosed { + logger.Info("Starting HTTP server", "addr", server.Addr) + if err := server.Serve(listener); err != nil && err != http.ErrServerClosed { logger.Error("HTTP server error", "error", err) } }() - return server + return server, nil } -// NewWatcher creates a new API Watcher instance -func NewWatcher(cfg *Config, logger *slog.Logger) (*Watcher, error) { - kubeconfig := cfg.KubeConfig - if kubeconfig == "" { - kubeconfig = os.Getenv("KUBECONFIG") - if kubeconfig == "" { - homeDir, err := os.UserHomeDir() - if err == nil { - kubeconfig = filepath.Join(homeDir, ".kube", "config") +// buildRestConfig resolves a kubeconfig path (explicit config, KUBECONFIG env +// var, then ~/.kube/config) and falls back to in-cluster configuration when +// no kubeconfig file is found +func buildRestConfig(kubeConfig string) (*rest.Config, error) { + if kubeConfig == "" { + kubeConfig = os.Getenv("KUBECONFIG") + } + if kubeConfig == "" { + if homeDir, err := os.UserHomeDir(); err == nil { + candidate := filepath.Join(homeDir, ".kube", "config") + if _, err := os.Stat(candidate); err == nil { + kubeConfig = candidate } } } - config, err := clientcmd.BuildConfigFromFlags("", kubeconfig) + if kubeConfig == "" { + config, err := rest.InClusterConfig() + if err != nil { + return nil, fmt.Errorf("no kubeconfig found and in-cluster config failed: %w", err) + } + return config, nil + } + + config, err := clientcmd.BuildConfigFromFlags("", kubeConfig) + if err != nil { + return nil, fmt.Errorf("build kubeconfig from %s: %w", kubeConfig, err) + } + return config, nil +} + +// NewWatcher creates a new API Watcher instance +func NewWatcher(cfg *Config, logger *slog.Logger) (*Watcher, error) { + config, err := buildRestConfig(cfg.KubeConfig) if err != nil { - return nil, fmt.Errorf("build kubeconfig: %w", err) + return nil, err } clientset, err := kubernetes.NewForConfig(config) @@ -310,14 +363,29 @@ func NewWatcher(cfg *Config, logger *slog.Logger) (*Watcher, error) { } return &Watcher{ - config: cfg, - client: clientset, - logger: logger, - nodeIPs: make(map[string]string), - tmpl: tmpl, + config: cfg, + client: clientset, + logger: logger, + nodeIPs: make(map[string]string), + tmpl: tmpl, + renderCh: make(chan struct{}, 1), + debounce: defaultDebounce, }, nil } +// nodeFromObj extracts a Node from an informer event object, unwrapping +// DeletedFinalStateUnknown tombstones delivered when a delete was missed. +func nodeFromObj(obj any) (*corev1.Node, bool) { + if node, ok := obj.(*corev1.Node); ok { + return node, true + } + if tombstone, ok := obj.(cache.DeletedFinalStateUnknown); ok { + node, ok := tombstone.Obj.(*corev1.Node) + return node, ok + } + return nil, false +} + // Run starts the watcher func (w *Watcher) Run(ctx context.Context) error { w.logger.Info("Starting node watcher") @@ -333,23 +401,36 @@ func (w *Watcher) Run(ctx context.Context) error { // - Handle node cordoning/draining? _, err := nodeInformer.AddEventHandler(cache.ResourceEventHandlerFuncs{ AddFunc: func(obj any) { - node := obj.(*corev1.Node) - w.handleNodeEvent("ADD", node) + if node, ok := nodeFromObj(obj); ok { + w.handleNodeEvent("ADD", node) + } else { + w.logger.Warn("Received non-node object in ADD event") + } }, UpdateFunc: func(oldObj, newObj any) { - node := newObj.(*corev1.Node) - w.handleNodeEvent("UPDATE", node) + if node, ok := nodeFromObj(newObj); ok { + w.handleNodeEvent("UPDATE", node) + } else { + w.logger.Warn("Received non-node object in UPDATE event") + } }, DeleteFunc: func(obj any) { - node := obj.(*corev1.Node) - w.handleNodeEvent("DELETE", node) + if node, ok := nodeFromObj(obj); ok { + w.handleNodeEvent("DELETE", node) + } else { + w.logger.Warn("Received non-node object in DELETE event") + } }, }) if err != nil { return fmt.Errorf("add event handler: %w", err) } - // Start informer + // Start the debounced render loop; node events only signal it + go w.renderLoop(ctx) + + // Start informer. It replays every existing node as an ADD event, + // so no separate initial sync is needed factory.Start(ctx.Done()) // Wait for cache sync @@ -358,73 +439,47 @@ func (w *Watcher) Run(ctx context.Context) error { return fmt.Errorf("failed to sync cache") } - w.logger.Info("Cache synced, performing initial sync") - - // Perform initial sync to get all current nodes - // This will not fail even if there are no nodes yet - if err := w.initialSync(nodeInformer); err != nil { - w.logger.Error("Initial sync failed, continuing to watch", "error", err) - } else { - w.logger.Info("Initial sync complete, watching for node changes") - } + w.logger.Info("Cache synced, watching for node changes") <-ctx.Done() return nil } -// initialSync fetches all current nodes and renders the initial template -func (w *Watcher) initialSync(informer cache.SharedIndexInformer) error { - w.mu.Lock() - defer w.mu.Unlock() - - // List all nodes from informer's store - items := informer.GetStore().List() - w.logger.Info("Initial node discovery", "count", len(items)) - - // Extract external IPs from all nodes - for _, item := range items { - node, ok := item.(*corev1.Node) - if !ok { - w.logger.Warn("Unexpected object type in store") - continue - } - var externalIP string - for _, addr := range node.Status.Addresses { - if addr.Type == corev1.NodeExternalIP { - externalIP = addr.Address - break +// renderLoop renders and executes after node changes, debouncing bursts of +// events into a single render +func (w *Watcher) renderLoop(ctx context.Context) { + timer := time.NewTimer(0) + if !timer.Stop() { + <-timer.C + } + pending := false + + for { + select { + case <-ctx.Done(): + timer.Stop() + return + case <-w.renderCh: + if !pending { + timer.Reset(w.debounce) + pending = true + } + case <-timer.C: + pending = false + if err := w.renderAndExecute(); err != nil { + w.logger.Error("Failed to render and execute", "error", err) } - } - - if externalIP != "" { - w.nodeIPs[node.Name] = externalIP - w.logger.Info("Discovered node", "node", node.Name, "ip", externalIP) - } else { - w.logger.Debug("Node has no external IP", "node", node.Name) } } +} - // Update node count gauge - currentNodeCount.Set(float64(len(w.nodeIPs))) - - // Check minimum node count (warning only, don't fail on startup) - if len(w.nodeIPs) < w.config.MinNodeCount { - w.logger.Warn("Node count below minimum, skipping initial render", - "current", len(w.nodeIPs), - "minimum", w.config.MinNodeCount, - ) - return nil - } - - // Render and execute for initial state - if len(w.nodeIPs) > 0 { - if err := w.renderAndExecute(); err != nil { - w.logger.Error("Initial render failed, will retry on node changes", "error", err) - // Don't return error - continue watching - } +// signalRender wakes the render loop without blocking; if a render is +// already pending the signal is coalesced +func (w *Watcher) signalRender() { + select { + case w.renderCh <- struct{}{}: + default: } - - return nil } // handleNodeEvent processes node events @@ -472,6 +527,12 @@ func (w *Watcher) handleNodeEvent(eventType string, node *corev1.Node) { w.logger.Info("Node IP changed", "node", nodeName, "oldIP", oldIP, "newIP", newIP) } } + } else if oldIP != "" { + // Safety net: a node that loses its external IP keeps its last + // known IP until the node object is deleted, so a transient + // cloud-provider hiccup never drops it from the rendered list + w.logger.Warn("Node lost its external IP, retaining last known IP until node is deleted", + "node", nodeName, "ip", oldIP) } // Update node count gauge @@ -483,33 +544,42 @@ func (w *Watcher) handleNodeEvent(eventType string, node *corev1.Node) { return } - // Safety check: prevent removing all nodes - if len(w.nodeIPs) < w.config.MinNodeCount { - w.logger.Error("Safety check failed: node count below minimum", - "current", len(w.nodeIPs), - "minimum", w.config.MinNodeCount, - ) - return - } - - // Render and execute - if err := w.renderAndExecute(); err != nil { - w.logger.Error("Failed to render and execute", "error", err) - } + // Wake the debounced render loop + w.signalRender() } -// renderAndExecute renders the template and executes the command +// renderAndExecute renders the template and executes the command. +// It snapshots state under the read lock so event handling is never +// blocked by file writes or command execution func (w *Watcher) renderAndExecute() error { - // Build node data + // Build node data from a snapshot of current state + w.mu.RLock() nodes := make([]NodeInfo, 0, len(w.nodeIPs)) - allIPs := make([]string, 0, len(w.nodeIPs)+len(w.config.StaticIPs)) - for name, ip := range w.nodeIPs { nodes = append(nodes, NodeInfo{ Name: name, ExternalIP: ip, }) - allIPs = append(allIPs, ip) + } + w.mu.RUnlock() + + // Sort by node name so the rendered output is deterministic + sort.Slice(nodes, func(i, j int) bool { + return nodes[i].Name < nodes[j].Name + }) + + allIPs := make([]string, 0, len(nodes)+len(w.config.StaticIPs)) + for _, node := range nodes { + allIPs = append(allIPs, node.ExternalIP) + } + + // Safety check: never render an empty/short node list + if len(nodes) < w.config.MinNodeCount { + w.logger.Warn("Safety check failed: node count below minimum, skipping render", + "current", len(nodes), + "minimum", w.config.MinNodeCount, + ) + return nil } // Add our static IPs @@ -529,24 +599,21 @@ func (w *Watcher) renderAndExecute() error { return nil } - // Render template to file + // Render template to a buffer first so a failed render never + // touches the existing output file w.logger.Info("Rendering template", "output", w.config.OutputPath, "nodeCount", len(nodes)) - outputFile, err := os.Create(w.config.OutputPath) - if err != nil { - rendersTotal.WithLabelValues("failure").Inc() - return fmt.Errorf("create output file: %w", err) - } - defer outputFile.Close() - - if err := w.tmpl.Execute(outputFile, data); err != nil { + var buf bytes.Buffer + if err := w.tmpl.Execute(&buf, data); err != nil { rendersTotal.WithLabelValues("failure").Inc() return fmt.Errorf("execute template: %w", err) } - if err := outputFile.Sync(); err != nil { + // Write to a temp file in the same directory and rename it over the + // output path so consumers never see a partial file + if err := w.writeOutputAtomic(buf.Bytes()); err != nil { rendersTotal.WithLabelValues("failure").Inc() - return fmt.Errorf("sync output file: %w", err) + return err } w.currentHash = dataHash @@ -556,6 +623,39 @@ func (w *Watcher) renderAndExecute() error { return w.executeCommand() } +// writeOutputAtomic writes content to a temp file in the output directory, +// fsyncs it and renames it over the output path +func (w *Watcher) writeOutputAtomic(content []byte) error { + dir := filepath.Dir(w.config.OutputPath) + tmpFile, err := os.CreateTemp(dir, ".node-ips-*.tmp") + if err != nil { + return fmt.Errorf("create temp output file: %w", err) + } + tmpName := tmpFile.Name() + defer os.Remove(tmpName) // no-op after successful rename + + if _, err := tmpFile.Write(content); err != nil { + tmpFile.Close() + return fmt.Errorf("write temp output file: %w", err) + } + if err := tmpFile.Chmod(0644); err != nil { + tmpFile.Close() + return fmt.Errorf("chmod temp output file: %w", err) + } + if err := tmpFile.Sync(); err != nil { + tmpFile.Close() + return fmt.Errorf("sync temp output file: %w", err) + } + if err := tmpFile.Close(); err != nil { + return fmt.Errorf("close temp output file: %w", err) + } + + if err := os.Rename(tmpName, w.config.OutputPath); err != nil { + return fmt.Errorf("rename temp output file: %w", err) + } + return nil +} + func (w *Watcher) calculateHash(data NodeData) string { h := sha256.New() @@ -566,11 +666,20 @@ func (w *Watcher) calculateHash(data NodeData) string { return nodes[i].Name < nodes[j].Name }) + // Separate fields with a NUL byte so distinct states never + // concatenate to the same byte stream + sep := []byte{0} + for _, node := range nodes { h.Write([]byte(node.Name)) + h.Write(sep) h.Write([]byte(node.ExternalIP)) + h.Write(sep) } + // Section marker between nodes and static IPs + h.Write([]byte{1}) + // Sort IPs for consistent hashing ips := make([]string, len(data.StaticIPs)) copy(ips, data.StaticIPs) @@ -578,6 +687,7 @@ func (w *Watcher) calculateHash(data NodeData) string { for _, ip := range ips { h.Write([]byte(ip)) + h.Write(sep) } return hex.EncodeToString(h.Sum(nil)) @@ -590,7 +700,10 @@ func (w *Watcher) executeCommand() error { "arg", w.config.OutputPath, ) - cmd := exec.Command(w.config.Command, w.config.OutputPath) + ctx, cancel := context.WithTimeout(context.Background(), time.Duration(w.config.CommandTimeout)*time.Second) + defer cancel() + + cmd := exec.CommandContext(ctx, w.config.Command, w.config.OutputPath) cmd.Stdout = os.Stdout cmd.Stderr = os.Stderr diff --git a/main_test.go b/main_test.go index 7ea989d..5b12757 100644 --- a/main_test.go +++ b/main_test.go @@ -1,15 +1,59 @@ package main import ( + "context" + "errors" + "fmt" "io" "log/slog" "net/http" "os" + "path/filepath" "strings" "testing" + "text/template" "time" + + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/client-go/rest" + "k8s.io/client-go/tools/cache" ) +func TestNodeFromObj(t *testing.T) { + node := &corev1.Node{ObjectMeta: metav1.ObjectMeta{Name: "node1"}} + + t.Run("plain node object", func(t *testing.T) { + got, ok := nodeFromObj(node) + if !ok || got != node { + t.Errorf("expected node back, got %v ok=%v", got, ok) + } + }) + + t.Run("tombstone wrapping a node", func(t *testing.T) { + tombstone := cache.DeletedFinalStateUnknown{Key: "node1", Obj: node} + got, ok := nodeFromObj(tombstone) + if !ok || got != node { + t.Errorf("expected node from tombstone, got %v ok=%v", got, ok) + } + }) + + t.Run("tombstone wrapping garbage", func(t *testing.T) { + tombstone := cache.DeletedFinalStateUnknown{Key: "x", Obj: "not a node"} + got, ok := nodeFromObj(tombstone) + if ok || got != nil { + t.Errorf("expected nil,false for garbage tombstone, got %v ok=%v", got, ok) + } + }) + + t.Run("unrelated type", func(t *testing.T) { + got, ok := nodeFromObj(42) + if ok || got != nil { + t.Errorf("expected nil,false for unrelated type, got %v ok=%v", got, ok) + } + }) +} + func TestCalculateHash(t *testing.T) { w := &Watcher{} @@ -141,6 +185,44 @@ func TestCalculateHash(t *testing.T) { } }) + t.Run("field boundaries affect hash (no concatenation collisions)", func(t *testing.T) { + // Without separators both hash the bytes "node110.0.0.1" + data1 := NodeData{ + Nodes: []NodeInfo{{Name: "node1", ExternalIP: "10.0.0.1"}}, + } + + data2 := NodeData{ + Nodes: []NodeInfo{{Name: "node11", ExternalIP: "0.0.0.1"}}, + } + + hash1 := w.calculateHash(data1) + hash2 := w.calculateHash(data2) + + if hash1 == hash2 { + t.Errorf("different data produced same hash (missing field separator): %s", hash1) + } + }) + + t.Run("node section and static IP section are distinct", func(t *testing.T) { + // A node entry must not collide with a static IP carrying the same bytes + data1 := NodeData{ + Nodes: []NodeInfo{{Name: "node1", ExternalIP: "1.2.3.4"}}, + StaticIPs: []string{}, + } + + data2 := NodeData{ + Nodes: []NodeInfo{}, + StaticIPs: []string{"node1", "1.2.3.4"}, + } + + hash1 := w.calculateHash(data1) + hash2 := w.calculateHash(data2) + + if hash1 == hash2 { + t.Errorf("node entry collided with static IPs: %s", hash1) + } + }) + t.Run("empty data produces consistent hash", func(t *testing.T) { data1 := NodeData{ Nodes: []NodeInfo{}, @@ -165,15 +247,460 @@ func TestCalculateHash(t *testing.T) { }) } +// newTestWatcher builds a Watcher suitable for unit tests, with the template +// parsed from a string and a no-op command. +func newTestWatcher(t *testing.T, tmplText string, cfg *Config) *Watcher { + t.Helper() + tmpl, err := template.New("test").Parse(tmplText) + if err != nil { + t.Fatalf("parse test template: %v", err) + } + if cfg.Command == "" { + cfg.Command = "true" + } + if cfg.MinNodeCount == 0 { + cfg.MinNodeCount = 1 + } + if cfg.CommandTimeout == 0 { + cfg.CommandTimeout = 30 + } + return &Watcher{ + config: cfg, + logger: slog.New(slog.NewTextHandler(io.Discard, nil)), + nodeIPs: map[string]string{"node1": "1.2.3.4"}, + tmpl: tmpl, + renderCh: make(chan struct{}, 1), + debounce: defaultDebounce, + } +} + +func TestRenderAndExecute(t *testing.T) { + t.Run("failed render preserves previous output file", func(t *testing.T) { + outputPath := filepath.Join(t.TempDir(), "out.conf") + previous := "previous good content" + if err := os.WriteFile(outputPath, []byte(previous), 0644); err != nil { + t.Fatal(err) + } + + w := newTestWatcher(t, "{{.NoSuchField}}", &Config{OutputPath: outputPath}) + + if err := w.renderAndExecute(); err == nil { + t.Fatal("expected render error, got nil") + } + + got, err := os.ReadFile(outputPath) + if err != nil { + t.Fatalf("read output file: %v", err) + } + if string(got) != previous { + t.Errorf("previous output destroyed by failed render: got %q", string(got)) + } + }) + + t.Run("successful render writes expected content", func(t *testing.T) { + dir := t.TempDir() + outputPath := filepath.Join(dir, "out.conf") + + w := newTestWatcher(t, "{{range .AllIPs}}{{.}}\n{{end}}", &Config{OutputPath: outputPath}) + + if err := w.renderAndExecute(); err != nil { + t.Fatalf("renderAndExecute: %v", err) + } + + got, err := os.ReadFile(outputPath) + if err != nil { + t.Fatalf("read output file: %v", err) + } + if string(got) != "1.2.3.4\n" { + t.Errorf("unexpected output content: %q", string(got)) + } + + info, err := os.Stat(outputPath) + if err != nil { + t.Fatal(err) + } + if info.Mode().Perm() != 0644 { + t.Errorf("expected mode 0644, got %v", info.Mode().Perm()) + } + + entries, err := os.ReadDir(dir) + if err != nil { + t.Fatal(err) + } + if len(entries) != 1 { + t.Errorf("expected only the output file in dir, found %d entries", len(entries)) + } + }) +} + +const testKubeconfig = `apiVersion: v1 +kind: Config +clusters: +- cluster: + server: https://127.0.0.1:6443 + name: test +contexts: +- context: + cluster: test + user: test + name: test +current-context: test +users: +- name: test + user: {} +` + +func TestBuildRestConfig(t *testing.T) { + t.Run("falls back to in-cluster config when no kubeconfig found", func(t *testing.T) { + t.Setenv("HOME", t.TempDir()) + t.Setenv("KUBECONFIG", "") + + _, err := buildRestConfig("") + if !errors.Is(err, rest.ErrNotInCluster) { + t.Errorf("expected in-cluster fallback (ErrNotInCluster outside a cluster), got: %v", err) + } + }) + + t.Run("uses explicit kubeconfig path", func(t *testing.T) { + path := filepath.Join(t.TempDir(), "kubeconfig") + if err := os.WriteFile(path, []byte(testKubeconfig), 0600); err != nil { + t.Fatal(err) + } + + cfg, err := buildRestConfig(path) + if err != nil { + t.Fatalf("buildRestConfig: %v", err) + } + if cfg.Host != "https://127.0.0.1:6443" { + t.Errorf("unexpected host: %s", cfg.Host) + } + }) + + t.Run("uses KUBECONFIG env var", func(t *testing.T) { + path := filepath.Join(t.TempDir(), "kubeconfig") + if err := os.WriteFile(path, []byte(testKubeconfig), 0600); err != nil { + t.Fatal(err) + } + t.Setenv("KUBECONFIG", path) + + cfg, err := buildRestConfig("") + if err != nil { + t.Fatalf("buildRestConfig: %v", err) + } + if cfg.Host != "https://127.0.0.1:6443" { + t.Errorf("unexpected host: %s", cfg.Host) + } + }) +} + +func TestExecuteCommandTimeout(t *testing.T) { + dir := t.TempDir() + script := filepath.Join(dir, "hang.sh") + if err := os.WriteFile(script, []byte("#!/bin/sh\nexec sleep 30\n"), 0755); err != nil { + t.Fatal(err) + } + + w := newTestWatcher(t, "x", &Config{ + OutputPath: filepath.Join(dir, "out.conf"), + Command: script, + CommandTimeout: 1, + }) + + start := time.Now() + err := w.executeCommand() + elapsed := time.Since(start) + + if err == nil { + t.Fatal("expected error from timed-out command, got nil") + } + if elapsed > 5*time.Second { + t.Fatalf("command did not time out promptly, took %v", elapsed) + } +} + +func testNode(name, externalIP string) *corev1.Node { + node := &corev1.Node{ObjectMeta: metav1.ObjectMeta{Name: name}} + if externalIP != "" { + node.Status.Addresses = []corev1.NodeAddress{ + {Type: corev1.NodeExternalIP, Address: externalIP}, + } + } + return node +} + +func TestDebouncedRender(t *testing.T) { + dir := t.TempDir() + countFile := filepath.Join(dir, "count") + script := filepath.Join(dir, "count.sh") + if err := os.WriteFile(script, []byte("#!/bin/sh\necho run >> "+countFile+"\n"), 0755); err != nil { + t.Fatal(err) + } + + w := newTestWatcher(t, "{{range .AllIPs}}{{.}}\n{{end}}", &Config{ + OutputPath: filepath.Join(dir, "out.conf"), + Command: script, + }) + w.nodeIPs = map[string]string{} + w.debounce = 50 * time.Millisecond + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + go w.renderLoop(ctx) + + // Burst of events (like informer replay at startup) must coalesce + // into a single render+command + for i := range 5 { + w.handleNodeEvent("ADD", testNode(fmt.Sprintf("node%d", i), fmt.Sprintf("10.0.0.%d", i))) + } + + time.Sleep(300 * time.Millisecond) + + data, err := os.ReadFile(countFile) + if err != nil { + t.Fatalf("command never ran: %v", err) + } + if got := strings.Count(string(data), "run"); got != 1 { + t.Errorf("expected exactly 1 command execution for event burst, got %d", got) + } + + // A later change triggers exactly one more render + w.handleNodeEvent("UPDATE", testNode("node0", "10.0.1.99")) + time.Sleep(300 * time.Millisecond) + + data, err = os.ReadFile(countFile) + if err != nil { + t.Fatal(err) + } + if got := strings.Count(string(data), "run"); got != 2 { + t.Errorf("expected 2 command executions after second change, got %d", got) + } +} + +func TestNodeLosesExternalIP(t *testing.T) { + var logBuf strings.Builder + + w := newTestWatcher(t, "x", &Config{OutputPath: filepath.Join(t.TempDir(), "out.conf")}) + w.logger = slog.New(slog.NewTextHandler(&logBuf, nil)) + w.nodeIPs = map[string]string{"node1": "1.2.3.4"} + + w.handleNodeEvent("UPDATE", testNode("node1", "")) + + // Safety net: the last known IP is retained so a transient cloud-provider + // hiccup never drops a node from the rendered list + if got := w.nodeIPs["node1"]; got != "1.2.3.4" { + t.Errorf("expected stale IP to be retained, got %q", got) + } + + // But the operator must be told about it + if !strings.Contains(logBuf.String(), "level=WARN") { + t.Errorf("expected a WARN log when a node loses its external IP, logs:\n%s", logBuf.String()) + } +} + +func TestLoadConfig(t *testing.T) { + writeConfig := func(t *testing.T, content string) string { + t.Helper() + path := filepath.Join(t.TempDir(), "config.yaml") + if err := os.WriteFile(path, []byte(content), 0644); err != nil { + t.Fatal(err) + } + return path + } + + valid := "templatePath: tmpl\noutputPath: out\ncommand: /bin/echo\n" + + t.Run("defaults are applied", func(t *testing.T) { + cfg, err := loadConfig(writeConfig(t, valid), "", "", "", "", "") + if err != nil { + t.Fatalf("loadConfig: %v", err) + } + if cfg.LogLevel != "info" || cfg.ResyncInterval != 300 || cfg.MinNodeCount != 1 || + cfg.MetricsAddr != "localhost:8089" || cfg.CommandTimeout != 30 { + t.Errorf("unexpected defaults: %+v", cfg) + } + }) + + t.Run("missing required fields fail", func(t *testing.T) { + for _, content := range []string{ + "outputPath: out\ncommand: c\n", // no templatePath + "templatePath: t\ncommand: c\n", // no outputPath + "templatePath: t\noutputPath: o\n", // no command + } { + if _, err := loadConfig(writeConfig(t, content), "", "", "", "", ""); err == nil { + t.Errorf("expected error for config %q, got nil", content) + } + } + }) + + t.Run("flag overrides beat file values", func(t *testing.T) { + cfg, err := loadConfig(writeConfig(t, valid), "debug", "/kc", "/tmpl2", "/out2", "0.0.0.0:9999") + if err != nil { + t.Fatalf("loadConfig: %v", err) + } + if cfg.LogLevel != "debug" || cfg.KubeConfig != "/kc" || cfg.TemplatePath != "/tmpl2" || + cfg.OutputPath != "/out2" || cfg.MetricsAddr != "0.0.0.0:9999" { + t.Errorf("flag overrides not applied: %+v", cfg) + } + }) + + t.Run("negative or zero numeric values are rejected", func(t *testing.T) { + for _, content := range []string{ + valid + "resyncInterval: -1\n", + valid + "minNodeCount: -1\n", + valid + "commandTimeout: 0\n", + valid + "commandTimeout: -5\n", + } { + if _, err := loadConfig(writeConfig(t, content), "", "", "", "", ""); err == nil { + t.Errorf("expected validation error for config %q, got nil", content) + } + } + }) + + t.Run("minNodeCount zero is allowed to disable the safety check", func(t *testing.T) { + cfg, err := loadConfig(writeConfig(t, valid+"minNodeCount: 0\n"), "", "", "", "", "") + if err != nil { + t.Fatalf("loadConfig: %v", err) + } + if cfg.MinNodeCount != 0 { + t.Errorf("expected minNodeCount 0, got %d", cfg.MinNodeCount) + } + }) +} + +func TestHandleNodeEvent(t *testing.T) { + newWatcher := func(t *testing.T) *Watcher { + w := newTestWatcher(t, "x", &Config{OutputPath: filepath.Join(t.TempDir(), "out.conf")}) + w.nodeIPs = map[string]string{} + return w + } + + signalled := func(w *Watcher) bool { + select { + case <-w.renderCh: + return true + default: + return false + } + } + + t.Run("ADD with external IP stores node and signals render", func(t *testing.T) { + w := newWatcher(t) + w.handleNodeEvent("ADD", testNode("node1", "1.2.3.4")) + if w.nodeIPs["node1"] != "1.2.3.4" { + t.Errorf("node not stored: %v", w.nodeIPs) + } + if !signalled(w) { + t.Error("expected render signal") + } + }) + + t.Run("ADD without external IP is ignored", func(t *testing.T) { + w := newWatcher(t) + w.handleNodeEvent("ADD", testNode("node1", "")) + if len(w.nodeIPs) != 0 { + t.Errorf("expected no nodes stored, got %v", w.nodeIPs) + } + if signalled(w) { + t.Error("expected no render signal") + } + }) + + t.Run("UPDATE with changed IP replaces it and signals", func(t *testing.T) { + w := newWatcher(t) + w.nodeIPs["node1"] = "1.2.3.4" + w.handleNodeEvent("UPDATE", testNode("node1", "5.6.7.8")) + if w.nodeIPs["node1"] != "5.6.7.8" { + t.Errorf("IP not updated: %v", w.nodeIPs) + } + if !signalled(w) { + t.Error("expected render signal") + } + }) + + t.Run("UPDATE with same IP does not signal", func(t *testing.T) { + w := newWatcher(t) + w.nodeIPs["node1"] = "1.2.3.4" + w.handleNodeEvent("UPDATE", testNode("node1", "1.2.3.4")) + if signalled(w) { + t.Error("expected no render signal for unchanged IP") + } + }) + + t.Run("DELETE removes node and signals", func(t *testing.T) { + w := newWatcher(t) + w.nodeIPs["node1"] = "1.2.3.4" + w.handleNodeEvent("DELETE", testNode("node1", "1.2.3.4")) + if len(w.nodeIPs) != 0 { + t.Errorf("node not removed: %v", w.nodeIPs) + } + if !signalled(w) { + t.Error("expected render signal") + } + }) + + t.Run("DELETE of unknown node does not signal", func(t *testing.T) { + w := newWatcher(t) + w.handleNodeEvent("DELETE", testNode("ghost", "1.2.3.4")) + if signalled(w) { + t.Error("expected no render signal") + } + }) +} + +func TestRenderSkippedBelowMinNodeCount(t *testing.T) { + outputPath := filepath.Join(t.TempDir(), "out.conf") + w := newTestWatcher(t, "x", &Config{OutputPath: outputPath, MinNodeCount: 2}) + + if err := w.renderAndExecute(); err != nil { + t.Fatalf("renderAndExecute: %v", err) + } + + if _, err := os.Stat(outputPath); !os.IsNotExist(err) { + t.Errorf("expected no output file below minNodeCount, stat err: %v", err) + } +} + +func TestRenderOutputIsSorted(t *testing.T) { + outputPath := filepath.Join(t.TempDir(), "out.conf") + w := newTestWatcher(t, "{{range .Nodes}}{{.Name}}={{.ExternalIP}}\n{{end}}{{range .AllIPs}}{{.}}\n{{end}}", &Config{ + OutputPath: outputPath, + StaticIPs: []string{"192.168.0.1"}, + }) + w.nodeIPs = map[string]string{ + "zeta": "10.0.0.3", "alpha": "10.0.0.1", "mike": "10.0.0.2", + "echo": "10.0.0.9", "kilo": "10.0.0.7", "bravo": "10.0.0.5", + "xray": "10.0.0.8", "golf": "10.0.0.6", + } + + if err := w.renderAndExecute(); err != nil { + t.Fatalf("renderAndExecute: %v", err) + } + + got, err := os.ReadFile(outputPath) + if err != nil { + t.Fatal(err) + } + + want := "alpha=10.0.0.1\nbravo=10.0.0.5\necho=10.0.0.9\ngolf=10.0.0.6\n" + + "kilo=10.0.0.7\nmike=10.0.0.2\nxray=10.0.0.8\nzeta=10.0.0.3\n" + + "10.0.0.1\n10.0.0.5\n10.0.0.9\n10.0.0.6\n10.0.0.7\n10.0.0.2\n10.0.0.8\n10.0.0.3\n" + + "192.168.0.1\n" + if string(got) != want { + t.Errorf("output not sorted by node name:\ngot:\n%s\nwant:\n%s", got, want) + } +} + func TestHTTPEndpoints(t *testing.T) { logger := slog.New(slog.NewTextHandler(os.Stdout, &slog.HandlerOptions{Level: slog.LevelError})) - addr := "localhost:18080" - server := startHTTPServer(addr, logger) + // Port 0 lets the kernel pick a free port; the listener is open before + // startHTTPServer returns, so no sleep is needed + server, err := startHTTPServer("localhost:0", logger) + if err != nil { + t.Fatalf("startHTTPServer: %v", err) + } defer server.Close() - - // Allow some time for the server to start - time.Sleep(100 * time.Millisecond) + addr := server.Addr t.Run("healthz endpoint returns 200 OK", func(t *testing.T) { resp, err := http.Get("http://" + addr + "/healthz")