diff --git a/go.mod b/go.mod index 2a64d5fc74..790a9ef2a2 100644 --- a/go.mod +++ b/go.mod @@ -163,6 +163,8 @@ require ( google.golang.org/protobuf v1.36.10 gopkg.in/natefinch/lumberjack.v2 v2.2.1 gopkg.in/yaml.v3 v3.0.1 + k8s.io/apimachinery v0.34.2 + k8s.io/client-go v0.34.2 modernc.org/sqlite v1.29.6 ) @@ -175,11 +177,15 @@ require ( github.com/cenkalti/backoff/v5 v5.0.3 // indirect github.com/daulet/tokenizers v1.22.1 // indirect github.com/elastic/elastic-transport-go/v8 v8.7.0 // indirect + github.com/emicklei/go-restful/v3 v3.12.2 // indirect github.com/emirpasic/gods v1.18.1 // indirect github.com/envoyproxy/go-control-plane/envoy v1.36.0 // indirect github.com/fxamacker/cbor/v2 v2.9.0 // indirect github.com/go-errors/errors v1.5.1 // indirect github.com/go-jose/go-jose/v4 v4.1.3 // indirect + github.com/go-openapi/jsonpointer v0.21.0 // indirect + github.com/go-openapi/jsonreference v0.20.2 // indirect + github.com/go-openapi/swag v0.23.0 // indirect github.com/go-viper/mapstructure/v2 v2.4.0 // indirect github.com/go-zeromq/goczmq/v4 v4.2.2 // indirect github.com/godbus/dbus v0.0.0-20190726142602-4481cbc300e2 // indirect @@ -189,6 +195,7 @@ require ( github.com/gomlx/gomlx v0.21.0 // indirect github.com/gomlx/gopjrt v0.7.3 // indirect github.com/gomlx/onnx-gomlx v0.2.4 // indirect + github.com/google/gnostic-models v0.7.0 // indirect github.com/gsterjov/go-libsecret v0.0.0-20161001094733-a6f4afe4910c // indirect github.com/hamba/avro/v2 v2.29.0 // indirect github.com/json-iterator/go v1.1.12 // indirect @@ -200,6 +207,7 @@ require ( github.com/modern-go/reflect2 v1.0.3-0.20250322232337-35a7c28c31ee // indirect github.com/munnerz/goautoneg v0.0.0-20191010083416-a7dc8b61c822 // indirect github.com/ncruces/go-strftime v0.1.9 // indirect + github.com/spf13/pflag v1.0.6 // indirect github.com/spiffe/go-spiffe/v2 v2.6.0 // indirect github.com/theparanoids/crypki v1.20.9 // indirect github.com/yalue/onnxruntime_go v1.21.0 // indirect @@ -207,16 +215,19 @@ require ( go.opentelemetry.io/otel/exporters/otlp/otlpmetric/otlpmetricgrpc v1.38.0 // indirect go.opentelemetry.io/otel/exporters/otlp/otlpmetric/otlpmetrichttp v1.38.0 // indirect go.yaml.in/yaml/v2 v2.4.2 // indirect + go.yaml.in/yaml/v3 v3.0.4 // indirect golang.org/x/exp v0.0.0-20250717185816-542afb5b7346 // indirect golang.org/x/telemetry v0.0.0-20251008203120-078029d740a8 // indirect - k8s.io/apimachinery v0.34.2 // indirect - k8s.io/client-go v0.34.2 // indirect + gopkg.in/evanphx/json-patch.v4 v4.12.0 // indirect + k8s.io/api v0.34.2 // 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-20251002143259-bc988d571ff4 // indirect modernc.org/gc/v3 v3.0.0-20240107210532-573471604cb6 // indirect sigs.k8s.io/json v0.0.0-20241014173422-cfa47c3a1cc8 // indirect sigs.k8s.io/randfill v1.0.0 // indirect sigs.k8s.io/structured-merge-diff/v6 v6.3.0 // indirect + sigs.k8s.io/yaml v1.6.0 // indirect ) require ( diff --git a/go.sum b/go.sum index a6a2446f36..9e643fca51 100644 --- a/go.sum +++ b/go.sum @@ -1026,6 +1026,8 @@ github.com/elastic/elastic-transport-go/v8 v8.7.0 h1:OgTneVuXP2uip4BA658Xi6Hfw+P github.com/elastic/elastic-transport-go/v8 v8.7.0/go.mod h1:YLHer5cj0csTzNFXoNQ8qhtGY1GTvSqPnKWKaqQE3Hk= github.com/elastic/go-elasticsearch/v9 v9.0.0 h1:krpgPeJ2lC8apkaw6B58gKDYJq5eUhP8AMwpPt01Q/U= github.com/elastic/go-elasticsearch/v9 v9.0.0/go.mod h1:2PB5YQPpY5tWbF65MRqzEXA31PZOdXCkloQSOZtU14I= +github.com/emicklei/go-restful/v3 v3.12.2 h1:DhwDP0vY3k8ZzE0RunuJy8GhNpPL6zqLkDf9B/a0/xU= +github.com/emicklei/go-restful/v3 v3.12.2/go.mod h1:6n3XBCmQQb25CM2LCACGz8ukIrRry+4bhvbpWn3mrbc= github.com/emicklei/proto v1.10.0 h1:pDGyFRVV5RvV+nkBK9iy3q67FBy9Xa7vwrOTE+g5aGw= github.com/emicklei/proto v1.10.0/go.mod h1:rn1FgRS/FANiZdD2djyH7TMA9jdRDcYQ9IEN9yvjX0A= github.com/emirpasic/gods v1.18.1 h1:FXtiHYKDGKCW2KzwZKx0iC0PQmdlorYgdFG9jPXJ1Bc= @@ -1107,6 +1109,14 @@ 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-ole/go-ole v1.3.0 h1:Dt6ye7+vXGIKZ7Xtk4s6/xVdGDQynvom7xCFEdWr6uE= github.com/go-ole/go-ole v1.3.0/go.mod h1:5LS6F96DhAwUc7C+1HLexzMXY1xGRSryjyPPKW6zv78= +github.com/go-openapi/jsonpointer v0.19.6/go.mod h1:osyAmYz/mB/C3I+WsTTSgw1ONzaLJoLCyoi6/zppojs= +github.com/go-openapi/jsonpointer v0.21.0 h1:YgdVicSA9vH5RiHs9TZW5oyafXZFc6+2Vc1rr/O9oNQ= +github.com/go-openapi/jsonpointer v0.21.0/go.mod h1:IUyH9l/+uyhIYQ/PXVA41Rexl+kOkAPDdXEYns6fzUY= +github.com/go-openapi/jsonreference v0.20.2 h1:3sVjiK66+uXK/6oQ8xgcRKcFgQ5KXa2KvnJRumpMGbE= +github.com/go-openapi/jsonreference v0.20.2/go.mod h1:Bl1zwGIM8/wsvqjsOQLJ/SH+En5Ap4rVB5KVcIDZG2k= +github.com/go-openapi/swag v0.22.3/go.mod h1:UzaqsxGiab7freDnrUUra0MwWfN/q7tE4j+VcZ0yl14= +github.com/go-openapi/swag v0.23.0 h1:vsEVJDUo2hPJ2tu0/Xc+4noaxyEffXNIs3cOULZ+GrE= +github.com/go-openapi/swag v0.23.0/go.mod h1:esZ8ITTYEsH1V2trKHjAN8Ai7xHb8RV+YSZ577vPjgQ= github.com/go-pdf/fpdf v0.5.0/go.mod h1:HzcnA+A23uwogo0tp9yU+l3V+KXhiESpt1PMayhOh5M= github.com/go-pdf/fpdf v0.6.0/go.mod h1:HzcnA+A23uwogo0tp9yU+l3V+KXhiESpt1PMayhOh5M= github.com/go-quicktest/qt v1.101.0 h1:O1K29Txy5P2OK0dGo59b7b0LR6wKfIhttaAhHUyn7eI= @@ -1117,6 +1127,8 @@ github.com/go-sql-driver/mysql v1.6.0/go.mod h1:DCzpHaOWr8IXmIStZouvnhqoel9Qv2LB github.com/go-sql-driver/mysql v1.7.1 h1:lUIinVbN1DY0xBg0eMOzmmtGoHwWBbvnWubQUrtU8EI= github.com/go-sql-driver/mysql v1.7.1/go.mod h1:OXbVy3sEdcQ2Doequ6Z5BW6fXNQTmx+9S1MCJN5yJMI= github.com/go-stack/stack v1.8.0/go.mod h1:v0f6uXyyMGvRgIKkXu+yp6POWl0qKG85gN/melR3HDY= +github.com/go-task/slim-sprig/v3 v3.0.0 h1:sUs3vkvUymDpBKi3qH1YSqBQk9+9D/8M2mN1vB6EwHI= +github.com/go-task/slim-sprig/v3 v3.0.0/go.mod h1:W848ghGpv3Qj3dhTPRyJypKRiqCdHZiAzKg9hl15HA8= github.com/go-test/deep v1.1.1 h1:0r/53hagsehfO4bzD2Pgr/+RgHqhmf+k1Bpse2cTu1U= github.com/go-test/deep v1.1.1/go.mod h1:5C2ZWiW0ErCdrYzpqxLbTX7MG14M9iiw8DgHncVwcsE= github.com/go-viper/mapstructure/v2 v2.4.0 h1:EBsztssimR/CONLSZZ04E8qAkxNYq4Qp9LvH92wZUgs= @@ -1208,6 +1220,8 @@ github.com/google/btree v1.0.0/go.mod h1:lNA+9X1NB3Zf8V7Ke586lFgjr2dZNuvo3lPJSGZ github.com/google/flatbuffers v2.0.8+incompatible/go.mod h1:1AeVuKshWv4vARoZatz6mlQ0JxURH0Kv5+zNeJKJCa8= github.com/google/flatbuffers v25.2.10+incompatible h1:F3vclr7C3HpB1k9mxCGRMXq6FdUalZ6H/pNX4FP1v0Q= github.com/google/flatbuffers v25.2.10+incompatible/go.mod h1:1AeVuKshWv4vARoZatz6mlQ0JxURH0Kv5+zNeJKJCa8= +github.com/google/gnostic-models v0.7.0 h1:qwTtogB15McXDaNqTZdzPJRHvaVJlAl+HVQnLmJEJxo= +github.com/google/gnostic-models v0.7.0/go.mod h1:whL5G0m6dmc5cPxKc5bdKdEN3UjI7OUGxBlw57miDrQ= github.com/google/go-cmp v0.2.0/go.mod h1:oXzfMopK8JAjlY9xF4vHSVASa0yLyX7SntLO5aqRK0M= github.com/google/go-cmp v0.3.0/go.mod h1:8QqcDgzrUqlUb/G2PQTWiueGozuR1884gddMywk6iLU= github.com/google/go-cmp v0.3.1/go.mod h1:8QqcDgzrUqlUb/G2PQTWiueGozuR1884gddMywk6iLU= @@ -1613,6 +1627,8 @@ github.com/olivere/elastic/v7 v7.0.32 h1:R7CXvbu8Eq+WlsLgxmKVKPox0oOwAE/2T9Si5Bn github.com/olivere/elastic/v7 v7.0.32/go.mod h1:c7PVmLe3Fxq77PIfY/bZmxY/TAamBhCzZ8xDOE09a9k= github.com/onsi/ginkgo v1.16.5 h1:8xi0RTUf59SOSfEtZMvwTvXYMzG4gV23XVHOZiXNtnE= github.com/onsi/ginkgo v1.16.5/go.mod h1:+E8gABHa3K6zRBolWtd+ROzc/U5bkGt0FwiG042wbpU= +github.com/onsi/ginkgo/v2 v2.21.0 h1:7rg/4f3rB88pb5obDgNZrNHrQ4e6WpjonchcpuBRnZM= +github.com/onsi/ginkgo/v2 v2.21.0/go.mod h1:7Du3c42kxCUegi0IImZ1wUQzMBVecgIHjR1C+NkhLQo= github.com/onsi/gomega v1.35.1 h1:Cwbd75ZBPxFSuZ6T+rN/WCb/gOc6YgFBXLlZLhC7Ds4= github.com/onsi/gomega v1.35.1/go.mod h1:PvZbdDc8J6XJEpDK4HCuRBm8a6Fzp9/DmhC9C7yFlog= github.com/opencontainers/go-digest v1.0.0 h1:apOUWs51W5PlhuyGyz9FCeeBIOUDA/6nW8Oi/yOhh5U= @@ -1987,6 +2003,8 @@ go.uber.org/zap v1.27.0 h1:aJMhYGrd5QSmlpLMr2MftRKl7t8J8PTZPA732ud/XR8= go.uber.org/zap v1.27.0/go.mod h1:GB2qFLM7cTU87MWRP2mPIjqfIDnGu+VIO4V/SdhGo2E= go.yaml.in/yaml/v2 v2.4.2 h1:DzmwEr2rDGHl7lsFgAHxmNz/1NlQ7xLIrlN2h5d1eGI= go.yaml.in/yaml/v2 v2.4.2/go.mod h1:081UH+NErpNdqlCXm3TtEran0rJZGxAYx9hb/ELlsPU= +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/crypto v0.0.0-20180904163835-0709b304e793/go.mod h1:6SG95UA2DQfeDnfUPMdvaQW0Q7yPrPDi9nlGo2tz2b4= golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w= golang.org/x/crypto v0.0.0-20190411191339-88737f569e3a/go.mod h1:WFFai1msRO1wXaEeE5yQxYXgSfI8pQAWXbQop6sCtWE= @@ -2700,6 +2718,8 @@ gopkg.in/check.v1 v1.0.0-20200902074654-038fdea0a05b/go.mod h1:Co6ibVJAznAaIkqp8 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= gopkg.in/errgo.v2 v2.1.0/go.mod h1:hNsd1EY+bozCKY1Ytp96fpM3vjJbqLJn88ws8XvfDNI= +gopkg.in/evanphx/json-patch.v4 v4.12.0 h1:n6jtcsulIzXPJaxegRbvFNNrZDjbij7ny3gmSPG+6V4= +gopkg.in/evanphx/json-patch.v4 v4.12.0/go.mod h1:p8EYWUEYMpynmqDbY58zCKCFZw8pRWMG4EsWvDvM72M= gopkg.in/inconshreveable/log15.v2 v2.0.0-20180818164646-67afb5ed74ec/go.mod h1:aPpfJ7XW+gOuirDoZ8gHhLh3kZ1B08FtV2bbmy7Jv3s= gopkg.in/inf.v0 v0.9.1 h1:73M5CoZyi3ZLMOyDlQh031Cx6N9NDJ2Vvfl76EDAgDc= gopkg.in/inf.v0 v0.9.1/go.mod h1:cWUDdTG/fYaXco+Dcufb5Vnc6Gp2YChqWtbxRZE0mXw= @@ -2739,12 +2759,16 @@ honnef.co/go/tools v0.0.1-2019.2.3/go.mod h1:a3bituU0lyd329TUQxRnasdCoJDkEUEAqEt honnef.co/go/tools v0.0.1-2020.1.3/go.mod h1:X/FiERA/W4tHapMX5mGpAtMSVEeEUOyHaw9vFzvIQ3k= honnef.co/go/tools v0.0.1-2020.1.4/go.mod h1:X/FiERA/W4tHapMX5mGpAtMSVEeEUOyHaw9vFzvIQ3k= honnef.co/go/tools v0.1.3/go.mod h1:NgwopIslSNH47DimFoV78dnkksY2EFtX0ajyb3K/las= +k8s.io/api v0.34.2 h1:fsSUNZhV+bnL6Aqrp6O7lMTy6o5x2C4XLjnh//8SLYY= +k8s.io/api v0.34.2/go.mod h1:MMBPaWlED2a8w4RSeanD76f7opUoypY8TFYkSM+3XHw= k8s.io/apimachinery v0.34.2 h1:zQ12Uk3eMHPxrsbUJgNF8bTauTVR2WgqJsTmwTE/NW4= k8s.io/apimachinery v0.34.2/go.mod h1:/GwIlEcWuTX9zKIg2mbw0LRFIsXwrfoVxn+ef0X13lw= k8s.io/client-go v0.34.2 h1:Co6XiknN+uUZqiddlfAjT68184/37PS4QAzYvQvDR8M= k8s.io/client-go v0.34.2/go.mod h1:2VYDl1XXJsdcAxw7BenFslRQX28Dxz91U9MWKjX97fE= 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-20250710124328-f3f2b991d03b h1:MloQ9/bdJyIu9lb1PzujOPolHyvO06MXG5TUIj2mNAA= +k8s.io/kube-openapi v0.0.0-20250710124328-f3f2b991d03b/go.mod h1:UZ2yyWbFTpuhSbFhv24aGNOdoRdJZgsIObGBUaYVsts= k8s.io/utils v0.0.0-20251002143259-bc988d571ff4 h1:SjGebBtkBqHFOli+05xYbK8YF1Dzkbzn+gDM4X9T4Ck= k8s.io/utils v0.0.0-20251002143259-bc988d571ff4/go.mod h1:OLgZIPagt7ERELqWJFomSt595RzquPNLL48iOWgYOg0= lukechampine.com/uint128 v1.1.1/go.mod h1:c4eWIwlEGaxC/+H1VguhU4PHXNWDCDMUlWdIWl2j1gk= diff --git a/internal/impl/kubernetes/auth.go b/internal/impl/kubernetes/auth.go new file mode 100644 index 0000000000..0bc9273838 --- /dev/null +++ b/internal/impl/kubernetes/auth.go @@ -0,0 +1,274 @@ +package kubernetes + +import ( + "context" + "errors" + "fmt" + "os" + "path/filepath" + "strings" + + "github.com/warpstreamlabs/bento/internal/filepath/ifs" + + "k8s.io/apimachinery/pkg/api/meta" + "k8s.io/client-go/discovery" + "k8s.io/client-go/discovery/cached/memory" + "k8s.io/client-go/dynamic" + "k8s.io/client-go/kubernetes" + "k8s.io/client-go/rest" + "k8s.io/client-go/restmapper" + "k8s.io/client-go/tools/clientcmd" + + "github.com/warpstreamlabs/bento/public/service" +) + +// AuthFields returns the config fields for Kubernetes authentication. +// These fields are shared across all Kubernetes components. +func AuthFields() []*service.ConfigField { + return []*service.ConfigField{ + service.NewBoolField("auto_auth"). + Description("Automatically detect authentication method. Tries in-cluster config first, then falls back to kubeconfig."). + Default(true), + service.NewStringField("kubeconfig"). + Description("Path to kubeconfig file. If empty, uses $KUBECONFIG (if set) or the default kubeconfig location (~/.kube/config)."). + Default(""). + Optional(), + service.NewStringField("kubeconfig_yaml"). + Description("Kubeconfig content as a YAML string. Use this to embed kubeconfig directly in your config (e.g., from a secret or environment variable) instead of referencing a file path."). + Default(""). + Secret(). + Optional(), + service.NewStringField("context"). + Description("Kubernetes context to use from kubeconfig. If empty, uses the current context."). + Default(""). + Optional(), + service.NewStringField("api_server"). + Description("Kubernetes API server URL. Only used when providing explicit credentials."). + Default(""). + Optional(). + Advanced(), + service.NewStringField("token"). + Description("Bearer token for authentication. Can be a service account token."). + Default(""). + Secret(). + Optional(). + Advanced(), + service.NewStringField("token_file"). + Description("Path to file containing bearer token."). + Default(""). + Optional(). + Advanced(), + service.NewStringField("ca_file"). + Description("Path to CA certificate file for verifying API server."). + Default(""). + Optional(). + Advanced(), + service.NewBoolField("insecure_skip_verify"). + Description("Skip TLS certificate verification. Not recommended for production."). + Default(false). + Advanced(), + service.NewFloatField("client_qps"). + Description("QPS limit for Kubernetes API client. 0 uses the client-go default."). + Default(0). + Advanced(), + service.NewIntField("client_burst"). + Description("Burst limit for Kubernetes API client. 0 uses the client-go default."). + Default(0). + Advanced(), + } +} + +// ClientSet contains the typed, dynamic, and discovery Kubernetes clients. +type ClientSet struct { + Typed kubernetes.Interface + Dynamic dynamic.Interface + Discovery discovery.DiscoveryInterface + Mapper meta.RESTMapper + Config *rest.Config +} + +// GetClientSet creates Kubernetes clients from parsed configuration. +func GetClientSet(ctx context.Context, conf *service.ParsedConfig, fs *service.FS) (*ClientSet, error) { + autoAuth, err := conf.FieldBool("auto_auth") + if err != nil { + return nil, fmt.Errorf("failed to parse auto_auth: %w", err) + } + + var config *rest.Config + + if autoAuth { + // Try in-cluster first + config, err = rest.InClusterConfig() + if err != nil { + // Fall back to kubeconfig + config, err = buildKubeconfigClient(conf) + if err != nil { + return nil, fmt.Errorf("auto auth failed: not running in cluster and kubeconfig not available: %w", err) + } + } + } else { + // Check for explicit credentials first + apiServer, _ := conf.FieldString("api_server") + if apiServer != "" { + config, err = buildExplicitClient(conf, fs) + } else { + config, err = buildKubeconfigClient(conf) + } + if err != nil { + return nil, err + } + } + + // Apply TLS settings + insecure, _ := conf.FieldBool("insecure_skip_verify") + if insecure { + config.Insecure = true + config.CAFile = "" + config.CAData = nil + } + + qps, err := conf.FieldFloat("client_qps") + if err != nil { + return nil, fmt.Errorf("failed to parse client_qps: %w", err) + } + if qps > 0 { + config.QPS = float32(qps) + } + + burst, err := conf.FieldInt("client_burst") + if err != nil { + return nil, fmt.Errorf("failed to parse client_burst: %w", err) + } + if burst > 0 { + config.Burst = burst + } + + // Create typed client + typedClient, err := kubernetes.NewForConfig(config) + if err != nil { + return nil, fmt.Errorf("failed to create kubernetes client: %w", err) + } + + // Create dynamic client for CRD support + dynamicClient, err := dynamic.NewForConfig(config) + if err != nil { + return nil, fmt.Errorf("failed to create dynamic client: %w", err) + } + + // Create cached discovery client and RESTMapper for GVR resolution + discoveryClient := typedClient.Discovery() + cachedDiscovery := memory.NewMemCacheClient(discoveryClient) + mapper := restmapper.NewDeferredDiscoveryRESTMapper(cachedDiscovery) + + return &ClientSet{ + Typed: typedClient, + Dynamic: dynamicClient, + Discovery: discoveryClient, + Mapper: mapper, + Config: config, + }, nil +} + +func buildKubeconfigClient(conf *service.ParsedConfig) (*rest.Config, error) { + kubeconfigYAML, _ := conf.FieldString("kubeconfig_yaml") + kubeconfigPath, _ := conf.FieldString("kubeconfig") + kubeContext, _ := conf.FieldString("context") + + // If raw kubeconfig YAML is provided, use it directly + if kubeconfigYAML != "" { + clientConfig, err := clientcmd.NewClientConfigFromBytes([]byte(kubeconfigYAML)) + if err != nil { + return nil, fmt.Errorf("failed to parse kubeconfig_yaml: %w", err) + } + // Apply context override if specified + if kubeContext != "" { + rawConfig, err := clientConfig.RawConfig() + if err != nil { + return nil, fmt.Errorf("failed to get raw config: %w", err) + } + rawConfig.CurrentContext = kubeContext + clientConfig = clientcmd.NewDefaultClientConfig(rawConfig, &clientcmd.ConfigOverrides{}) + } + return clientConfig.ClientConfig() + } + + // Fall back to file-based kubeconfig + loadingRules := clientcmd.NewDefaultClientConfigLoadingRules() + if kubeconfigPath != "" { + loadingRules.ExplicitPath = expandHomePath(kubeconfigPath) + } else if envKubeconfig := os.Getenv("KUBECONFIG"); envKubeconfig != "" { + paths := filepath.SplitList(envKubeconfig) + expanded := make([]string, 0, len(paths)) + for _, p := range paths { + if p == "" { + continue + } + expanded = append(expanded, expandHomePath(p)) + } + if len(expanded) > 0 { + loadingRules.Precedence = expanded + } + } + + configOverrides := &clientcmd.ConfigOverrides{} + if kubeContext != "" { + configOverrides.CurrentContext = kubeContext + } + + clientConfig := clientcmd.NewNonInteractiveDeferredLoadingClientConfig( + loadingRules, + configOverrides, + ) + + return clientConfig.ClientConfig() +} + +func expandHomePath(path string) string { + if path == "~" { + if home, err := os.UserHomeDir(); err == nil && home != "" { + return home + } + } + + prefix := "~" + string(os.PathSeparator) + if strings.HasPrefix(path, prefix) { + if home, err := os.UserHomeDir(); err == nil && home != "" { + return filepath.Join(home, path[len(prefix):]) + } + } + + return path +} + +func buildExplicitClient(conf *service.ParsedConfig, fs *service.FS) (*rest.Config, error) { + apiServer, _ := conf.FieldString("api_server") + if apiServer == "" { + return nil, errors.New("api_server is required for explicit authentication") + } + + config := &rest.Config{ + Host: apiServer, + } + + // Token authentication + token, _ := conf.FieldString("token") + tokenFile, _ := conf.FieldString("token_file") + + if token != "" { + config.BearerToken = token + } else if tokenFile != "" { + tokenBytes, err := ifs.ReadFile(fs, tokenFile) + if err != nil { + return nil, fmt.Errorf("failed to read token file: %w", err) + } + config.BearerToken = strings.TrimSpace(string(tokenBytes)) + } + + // CA certificate + caFile, _ := conf.FieldString("ca_file") + if caFile != "" { + config.CAFile = caFile + } + + return config, nil +} diff --git a/internal/impl/kubernetes/auth_test.go b/internal/impl/kubernetes/auth_test.go new file mode 100644 index 0000000000..226b9ef677 --- /dev/null +++ b/internal/impl/kubernetes/auth_test.go @@ -0,0 +1,82 @@ +package kubernetes + +import ( + "fmt" + "os" + "path/filepath" + "testing" + + "github.com/stretchr/testify/require" + "github.com/warpstreamlabs/bento/public/service" +) + +func TestBuildKubeconfigUsesEnv(t *testing.T) { + dir := t.TempDir() + + kubeconfigPath := writeKubeconfig(t, dir, "https://example-env.invalid") + t.Setenv("KUBECONFIG", kubeconfigPath) + + conf := parseAuthConfig(t, "") + + clientCfg, err := buildKubeconfigClient(conf) + require.NoError(t, err) + require.Equal(t, "https://example-env.invalid", clientCfg.Host) +} + +func TestBuildKubeconfigExplicitOverridesEnv(t *testing.T) { + dir := t.TempDir() + + envKubeconfigPath := writeKubeconfig(t, dir, "https://example-env.invalid") + explicitKubeconfigPath := writeKubeconfig(t, dir, "https://example-explicit.invalid") + + t.Setenv("KUBECONFIG", envKubeconfigPath) + + conf := parseAuthConfig(t, fmt.Sprintf("kubeconfig: %s\n", explicitKubeconfigPath)) + + clientCfg, err := buildKubeconfigClient(conf) + require.NoError(t, err) + require.Equal(t, "https://example-explicit.invalid", clientCfg.Host) +} + +func parseAuthConfig(t *testing.T, yamlConfig string) *service.ParsedConfig { + spec := service.NewConfigSpec().Fields(AuthFields()...) + conf, err := spec.ParseYAML(yamlConfig, nil) + require.NoError(t, err) + return conf +} + +func writeKubeconfig(t *testing.T, dir, server string) string { + path := filepath.Join(dir, fmt.Sprintf("kubeconfig-%s.yaml", sanitizeServer(server))) + data := fmt.Sprintf(`apiVersion: v1 +kind: Config +clusters: +- name: test + cluster: + server: %s +contexts: +- name: test + context: + cluster: test + user: test +current-context: test +users: +- name: test + user: + token: dummy +`, server) + require.NoError(t, os.WriteFile(path, []byte(data), 0o644)) + return path +} + +func sanitizeServer(server string) string { + out := make([]byte, 0, len(server)) + for i := 0; i < len(server); i++ { + c := server[i] + if (c >= 'a' && c <= 'z') || (c >= 'A' && c <= 'Z') || (c >= '0' && c <= '9') { + out = append(out, c) + } else { + out = append(out, '_') + } + } + return string(out) +} diff --git a/internal/impl/kubernetes/auth_unix.go b/internal/impl/kubernetes/auth_unix.go new file mode 100644 index 0000000000..94235ab261 --- /dev/null +++ b/internal/impl/kubernetes/auth_unix.go @@ -0,0 +1,20 @@ +//go:build unix + +package kubernetes + +import ( + "strings" + + "github.com/warpstreamlabs/bento/internal/filepath/ifs" +) + +// InClusterNamespace returns the namespace this pod is running in, +// or "default" if not running in a cluster. +func InClusterNamespace() string { + // Try to read the namespace from the service account + nsBytes, err := ifs.ReadFile(ifs.OS(), "/var/run/secrets/kubernetes.io/serviceaccount/namespace") + if err == nil { + return strings.TrimSpace(string(nsBytes)) + } + return "default" +} diff --git a/internal/impl/kubernetes/auth_windows.go b/internal/impl/kubernetes/auth_windows.go new file mode 100644 index 0000000000..03f6a8728b --- /dev/null +++ b/internal/impl/kubernetes/auth_windows.go @@ -0,0 +1,9 @@ +//go:build windows + +package kubernetes + +// InClusterNamespace returns "default" on Windows as in-cluster +// authentication is not supported. +func InClusterNamespace() string { + return "default" +} diff --git a/internal/impl/kubernetes/config.go b/internal/impl/kubernetes/config.go new file mode 100644 index 0000000000..290af07029 --- /dev/null +++ b/internal/impl/kubernetes/config.go @@ -0,0 +1,68 @@ +package kubernetes + +import ( + "strings" + + "github.com/warpstreamlabs/bento/public/service" +) + +// CommonFields returns config fields shared across all Kubernetes inputs. +func CommonFields() []*service.ConfigField { + return []*service.ConfigField{ + service.NewStringListField("namespaces"). + Description("Namespaces to watch. Empty list means all namespaces."). + Default([]any{}). + Example([]string{"default"}). + Example([]string{"production", "staging"}), + service.NewStringMapField("label_selector"). + Description("Kubernetes label selector to filter resources."). + Default(map[string]any{}). + Example(map[string]any{"app": "myapp"}). + Example(map[string]any{"app": "myapp", "env": "prod"}), + service.NewStringMapField("field_selector"). + Description("Kubernetes field selector to filter resources."). + Default(map[string]any{}). + Example(map[string]any{"status.phase": "Running"}). + Example(map[string]any{"metadata.name": "my-pod"}). + Optional(). + Advanced(), + service.NewStringField("request_timeout"). + Description("Timeout for Kubernetes API requests such as list calls. Use \"0s\" to disable."). + Default("30s"). + Advanced(), + } +} + +// LabelSelectorFromMap converts a map of key-value pairs to a Kubernetes +// label selector string (e.g., "app=myapp,env=prod"). +func LabelSelectorFromMap(labels map[string]string) string { + if len(labels) == 0 { + return "" + } + pairs := make([]string, 0, len(labels)) + for k, v := range labels { + pairs = append(pairs, k+"="+v) + } + return strings.Join(pairs, ",") +} + +// metadataDescription returns the standard metadata documentation block. +func metadataDescription(fields ...string) string { + result := ` + +### Metadata + +This input adds the following metadata fields to each message: + +` + "```text" + ` +` + for _, f := range fields { + result += "- " + f + "\n" + } + result += "```" + ` + +You can access these metadata fields using +[function interpolation](/docs/configuration/interpolation#bloblang-queries). +` + return result +} diff --git a/internal/impl/kubernetes/input_watch.go b/internal/impl/kubernetes/input_watch.go new file mode 100644 index 0000000000..1b75e217a3 --- /dev/null +++ b/internal/impl/kubernetes/input_watch.go @@ -0,0 +1,523 @@ +package kubernetes + +import ( + "context" + "encoding/json" + "errors" + "fmt" + "net/http" + "sync" + "time" + + "github.com/Jeffail/shutdown" + "github.com/cenkalti/backoff/v4" + k8serrors "k8s.io/apimachinery/pkg/api/errors" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" + "k8s.io/apimachinery/pkg/runtime/schema" + "k8s.io/apimachinery/pkg/watch" + "k8s.io/client-go/dynamic" + + "github.com/warpstreamlabs/bento/internal/retries" + "github.com/warpstreamlabs/bento/public/service" +) + +func kubernetesWatchInputConfig() *service.ConfigSpec { + return service.NewConfigSpec(). + Stable(). + Categories("Services", "Kubernetes"). + Version("1.15.0"). + Summary("Watches Kubernetes resources for changes, similar to `kubectl get --watch`."). + Description(` +This input watches Kubernetes resources and emits events when resources +are added, modified, or deleted. It uses the Kubernetes discovery API +to automatically resolve resource types, supporting both standard +resources and Custom Resource Definitions (CRDs). + +### Watching Resources + +Specify any resource name recognized by the cluster: +` + "```yaml" + ` +input: + kubernetes_watch: + resource: pods +` + "```" + ` + +The resource name is resolved dynamically via the Kubernetes API, +so any valid resource type is supported (pods, deployments, services, +configmaps, or any installed CRD). + +### Explicit GVR for Custom Resources + +For custom resources where you need to specify the exact group/version: +` + "```yaml" + ` +input: + kubernetes_watch: + custom_resource: + group: stable.example.com + version: v1 + resource: crontabs +` + "```" + ` + +### Watch Events + +Each message includes the watch event type in metadata: +- **ADDED**: Resource was created +- **MODIFIED**: Resource was updated +- **DELETED**: Resource was removed +` + metadataDescription( + "kubernetes_watch_event_type", + "kubernetes_resource_kind", + "kubernetes_resource_name", + "kubernetes_resource_namespace", + "kubernetes_resource_version", + "kubernetes_resource_uid", + "kubernetes_resource_creation_timestamp", + ) + ` + +Additionally, all resource labels are added as metadata with the prefix ` + "`kubernetes_labels_`" + `, and all annotations are added with the prefix ` + "`kubernetes_annotations_`" + `. For example, a label ` + "`app: myapp`" + ` becomes metadata key ` + "`kubernetes_labels_app`" + `. +`). + Fields(AuthFields()...). + Fields(CommonFields()...). + Field(service.NewStringField("resource"). + Description("Standard Kubernetes resource type to watch."). + Default(""). + Example("pods"). + Example("deployments"). + Example("configmaps")). + Field(service.NewObjectField("custom_resource", + service.NewStringField("group"). + Description("API group for the custom resource (e.g., 'stable.example.com')."). + Default(""), + service.NewStringField("version"). + Description("API version for the custom resource (e.g., 'v1', 'v1beta1')."). + Default("v1"), + service.NewStringField("resource"). + Description("Plural name of the custom resource (e.g., 'crontabs')."). + Default(""), + ). + Description("Custom Resource Definition to watch. Use this for CRDs instead of 'resource'."). + Optional(). + Advanced()). + Field(service.NewStringListField("event_types"). + Description("Watch event types to include. Valid values are `ADDED`, `MODIFIED`, and `DELETED`."). + Default([]any{"ADDED", "MODIFIED", "DELETED"}). + Example([]string{"ADDED", "DELETED"}). + Example([]string{"MODIFIED"}). + LintRule(`root = this.filter(v -> !["ADDED", "MODIFIED", "DELETED"].contains(v)).map_each(v -> "invalid event type %q, must be one of: ADDED, MODIFIED, DELETED".format(v))`)). + Field(service.NewBoolField("include_initial_list"). + Description("Emit ADDED events for all existing resources when starting."). + Default(true)). + Fields(retries.CommonRetryBackOffFields(0, "1s", "60s", "0s")...). + LintRule(` + let has_resource = this.resource.or("") != "" + let has_custom = this.custom_resource.resource.or("") != "" + root = if !$has_resource && !$has_custom { + "either resource or custom_resource.resource must be specified" + } else if $has_resource && $has_custom { + "cannot specify both resource and custom_resource" + } + `) +} + +func init() { + err := service.RegisterInput( + "kubernetes_watch", kubernetesWatchInputConfig(), + func(conf *service.ParsedConfig, mgr *service.Resources) (service.Input, error) { + return newKubernetesWatchInput(conf, mgr) + }) + if err != nil { + panic(err) + } +} + +type watchEvent struct { + eventType string + object *unstructured.Unstructured +} + +type kubernetesWatchInput struct { + clientSet *ClientSet + log *service.Logger + + // Configuration + namespaces []string + labelSelector string + fieldSelector string + gvr schema.GroupVersionResource + eventTypes map[string]struct{} + includeInitialList bool + requestTimeout time.Duration + backoffCtor func() backoff.BackOff + + // State + mu sync.RWMutex + eventChan chan watchEvent + shutSig *shutdown.Signaller + resourceVers map[string]string + wg sync.WaitGroup +} + +func newKubernetesWatchInput(conf *service.ParsedConfig, mgr *service.Resources) (*kubernetesWatchInput, error) { + k := &kubernetesWatchInput{ + log: mgr.Logger(), + eventChan: make(chan watchEvent, 1000), + shutSig: shutdown.NewSignaller(), + resourceVers: make(map[string]string), + } + + var err error + + // Parse namespaces + if k.namespaces, err = conf.FieldStringList("namespaces"); err != nil { + return nil, err + } + + // Parse selectors + labelSelectorMap, err := conf.FieldStringMap("label_selector") + if err != nil { + return nil, err + } + k.labelSelector = LabelSelectorFromMap(labelSelectorMap) + fieldSelectorMap, err := conf.FieldStringMap("field_selector") + if err != nil { + return nil, err + } + k.fieldSelector = LabelSelectorFromMap(fieldSelectorMap) + + // Parse event types filter + eventTypesList, err := conf.FieldStringList("event_types") + if err != nil { + return nil, err + } + k.eventTypes = make(map[string]struct{}) + for _, et := range eventTypesList { + k.eventTypes[et] = struct{}{} + } + + // Parse behavior options + if k.includeInitialList, err = conf.FieldBool("include_initial_list"); err != nil { + return nil, err + } + requestTimeoutStr, err := conf.FieldString("request_timeout") + if err != nil { + return nil, err + } + if k.requestTimeout, err = time.ParseDuration(requestTimeoutStr); err != nil { + return nil, fmt.Errorf("failed to parse request_timeout: %w", err) + } + + // Parse backoff configuration + if k.backoffCtor, err = retries.CommonRetryBackOffCtorFromParsed(conf); err != nil { + return nil, fmt.Errorf("failed to parse backoff config: %w", err) + } + + // Get Kubernetes client (needed for RESTMapper) + if k.clientSet, err = GetClientSet(context.Background(), conf, mgr.FS()); err != nil { + return nil, fmt.Errorf("failed to create kubernetes client: %w", err) + } + + // Determine GVR from resource or custom_resource + resource, _ := conf.FieldString("resource") + if resource != "" { + // Use RESTMapper to dynamically resolve the GVR + gvr, err := k.clientSet.Mapper.ResourceFor(schema.GroupVersionResource{Resource: resource}) + if err != nil { + return nil, fmt.Errorf("failed to resolve resource %q: %w", resource, err) + } + k.gvr = gvr + } else { + // Custom resource with explicit GVR + crConf := conf.Namespace("custom_resource") + group, _ := crConf.FieldString("group") + version, _ := crConf.FieldString("version") + crResource, _ := crConf.FieldString("resource") + + if crResource == "" { + return nil, errors.New("custom_resource.resource is required when using custom_resource") + } + + k.gvr = schema.GroupVersionResource{ + Group: group, + Version: version, + Resource: crResource, + } + } + + return k, nil +} + +func (k *kubernetesWatchInput) getResourceVersion(namespace string) string { + k.mu.Lock() + defer k.mu.Unlock() + return k.resourceVers[namespace] +} + +func (k *kubernetesWatchInput) setResourceVersion(namespace, version string) { + k.mu.Lock() + defer k.mu.Unlock() + if version == "" { + delete(k.resourceVers, namespace) + return + } + k.resourceVers[namespace] = version +} + +func (k *kubernetesWatchInput) Connect(ctx context.Context) error { + // Start watch loops for each namespace + namespaces := k.namespaces + if len(namespaces) == 0 { + namespaces = []string{""} // Empty string = all namespaces + } + + for _, ns := range namespaces { + k.wg.Go(func() { + k.watchNamespace(ns) + }) + } + + return nil +} + +func (k *kubernetesWatchInput) watchNamespace(namespace string) { + dynamicClient := k.clientSet.Dynamic + boff := k.backoffCtor() + + // Initialize resource interface once (it only depends on namespace, which is constant) + var resourceInterface dynamic.ResourceInterface + if namespace == "" { + resourceInterface = dynamicClient.Resource(k.gvr) + } else { + resourceInterface = dynamicClient.Resource(k.gvr).Namespace(namespace) + } + + for { + select { + case <-k.shutSig.SoftStopChan(): + return + default: + } + + listOpts := metav1.ListOptions{ + LabelSelector: k.labelSelector, + FieldSelector: k.fieldSelector, + } + + // If include_initial_list is true on first run, list existing resources + resourceVersion := k.getResourceVersion(namespace) + if k.includeInitialList && resourceVersion == "" { + k.listExistingResources(resourceInterface, namespace) + } + + // Use stored resource version if available + resourceVersion = k.getResourceVersion(namespace) + if resourceVersion != "" { + listOpts.ResourceVersion = resourceVersion + } + + // Start watching + shouldExit := false + func() { + watchCtx, watchDone := k.shutSig.SoftStopCtx(context.Background()) + defer watchDone() // Ensure context is cancelled to avoid leaks + watcher, err := resourceInterface.Watch(watchCtx, listOpts) + if err != nil { + // If the error is a "410 Gone" error, it means the resource version is too old. + // Reset the resource version to force a fresh list on the next loop. + if serr, ok := err.(*k8serrors.StatusError); ok && serr.ErrStatus.Code == http.StatusGone { + k.log.Warnf("Watch for %s in namespace %s returned 410 Gone, resetting resource version", k.gvr.Resource, namespace) + k.setResourceVersion(namespace, "") + } else { + k.log.Errorf("Failed to watch %s in namespace %s: %v", k.gvr.Resource, namespace, err) + } + wait := boff.NextBackOff() + if wait == backoff.Stop { + k.log.Errorf("Max retries exceeded for watch %s in namespace %s", k.gvr.Resource, namespace) + shouldExit = true + return + } + select { + case <-time.After(wait): + case <-k.shutSig.SoftStopChan(): + shouldExit = true + return + } + return + } + + // Reset backoff on successful connection + boff.Reset() + k.processWatchEvents(watcher, namespace) + watcher.Stop() + }() + if shouldExit { + return + } + } +} + +func (k *kubernetesWatchInput) listExistingResources(resourceInterface dynamic.ResourceInterface, namespace string) { + listOpts := metav1.ListOptions{ + LabelSelector: k.labelSelector, + FieldSelector: k.fieldSelector, + Limit: 500, // Process in chunks of 500 + } + + for { + softCtx, done := k.shutSig.SoftStopCtx(context.Background()) + listCtx := softCtx + var cancel context.CancelFunc + if k.requestTimeout > 0 { + listCtx, cancel = context.WithTimeout(softCtx, k.requestTimeout) + } + list, err := resourceInterface.List(listCtx, listOpts) + if cancel != nil { + cancel() + } + done() + if err != nil { + k.log.Errorf("Failed to list %s: %v", k.gvr.Resource, err) + return + } + + // Store resource version for watch (update with latest) + k.setResourceVersion(namespace, list.GetResourceVersion()) + + // Check if ADDED events are filtered + if _, ok := k.eventTypes["ADDED"]; ok { + for i := range list.Items { + item := &list.Items[i] + select { + case k.eventChan <- watchEvent{eventType: "ADDED", object: item}: + case <-k.shutSig.SoftStopChan(): + return + } + } + } + + // Check if there are more items + continueToken := list.GetContinue() + if continueToken == "" { + break + } + listOpts.Continue = continueToken + } +} + +func (k *kubernetesWatchInput) processWatchEvents(watcher watch.Interface, namespace string) { + for { + select { + case <-k.shutSig.SoftStopChan(): + return + case event, ok := <-watcher.ResultChan(): + if !ok { + // Watch channel closed + return + } + + switch event.Type { + case watch.Added, watch.Modified, watch.Deleted: + obj, ok := event.Object.(*unstructured.Unstructured) + if !ok { + continue + } + + // Store resource version + k.setResourceVersion(namespace, obj.GetResourceVersion()) + + // Check event type filter + eventType := string(event.Type) + if _, ok := k.eventTypes[eventType]; !ok { + continue + } + + select { + case k.eventChan <- watchEvent{eventType: eventType, object: obj}: + case <-k.shutSig.SoftStopChan(): + return + } + + case watch.Error: + if status, ok := event.Object.(*metav1.Status); ok && status.Code == http.StatusGone { + k.log.Warnf("Watch for %s in namespace %s returned 410 Gone, resetting resource version", k.gvr.Resource, namespace) + k.setResourceVersion(namespace, "") + return + } + k.log.Errorf("Watch error for %s in namespace %s: %v", + k.gvr.Resource, namespace, event.Object) + return + } + } + } +} + +func (k *kubernetesWatchInput) Read(ctx context.Context) (*service.Message, service.AckFunc, error) { + select { + case event, ok := <-k.eventChan: + if !ok { + return nil, nil, service.ErrEndOfInput + } + return k.eventToMessage(event) + case <-ctx.Done(): + return nil, nil, ctx.Err() + case <-k.shutSig.SoftStopChan(): + return nil, nil, service.ErrEndOfInput + } +} + +func (k *kubernetesWatchInput) eventToMessage(event watchEvent) (*service.Message, service.AckFunc, error) { + obj := event.object + + // Serialize object to JSON + objJSON, err := json.Marshal(obj.Object) + if err != nil { + return nil, nil, fmt.Errorf("failed to marshal object: %w", err) + } + + msg := service.NewMessage(objJSON) + + // Add metadata + msg.MetaSetMut("kubernetes_watch_event_type", event.eventType) + msg.MetaSetMut("kubernetes_resource_kind", obj.GetKind()) + msg.MetaSetMut("kubernetes_resource_name", obj.GetName()) + msg.MetaSetMut("kubernetes_resource_namespace", obj.GetNamespace()) + msg.MetaSetMut("kubernetes_resource_version", obj.GetResourceVersion()) + msg.MetaSetMut("kubernetes_resource_uid", string(obj.GetUID())) + + creationTS := obj.GetCreationTimestamp() + if !creationTS.Time.IsZero() { + msg.MetaSetMut("kubernetes_resource_creation_timestamp", + creationTS.Format(time.RFC3339)) + } + + // Add labels as metadata + for key, value := range obj.GetLabels() { + msg.MetaSetMut("kubernetes_labels_"+key, value) + } + + // Add annotations as metadata + annotations := obj.GetAnnotations() + for key, value := range annotations { + msg.MetaSetMut("kubernetes_annotations_"+key, value) + } + + return msg, func(ctx context.Context, err error) error { + return nil // Watch events don't require acknowledgment + }, nil +} + +func (k *kubernetesWatchInput) Close(ctx context.Context) error { + go func() { + k.shutSig.TriggerSoftStop() + k.wg.Wait() + close(k.eventChan) + k.shutSig.TriggerHasStopped() + }() + + select { + case <-ctx.Done(): + return ctx.Err() + case <-k.shutSig.HasStoppedChan(): + return nil + } +} diff --git a/internal/impl/kubernetes/input_watch_test.go b/internal/impl/kubernetes/input_watch_test.go new file mode 100644 index 0000000000..2fdb132389 --- /dev/null +++ b/internal/impl/kubernetes/input_watch_test.go @@ -0,0 +1,263 @@ +package kubernetes + +import ( + "context" + "sync" + "testing" + "time" + + "github.com/Jeffail/shutdown" + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" + "github.com/warpstreamlabs/bento/public/service" + "k8s.io/apimachinery/pkg/apis/meta/v1/unstructured" +) + +func TestKubernetesWatchConfigParse(t *testing.T) { + spec := kubernetesWatchInputConfig() + + tests := []struct { + name string + config string + expectError bool + }{ + { + name: "watch pods", + config: ` +resource: pods +`, + expectError: false, + }, + { + name: "watch deployments in namespace", + config: ` +resource: deployments +namespaces: + - production +label_selector: + app: backend +`, + expectError: false, + }, + { + name: "watch custom resource", + config: ` +custom_resource: + group: stable.example.com + version: v1 + resource: crontabs +`, + expectError: false, + }, + { + name: "filter event types", + config: ` +resource: pods +event_types: + - ADDED + - DELETED +include_initial_list: false +`, + expectError: false, + }, + { + name: "all namespaces with resource", + config: ` +resource: services +namespaces: [] +`, + expectError: false, + }, + } + + for _, tt := range tests { + t.Run(tt.name, func(t *testing.T) { + conf, err := spec.ParseYAML(tt.config, nil) + if tt.expectError { + require.Error(t, err) + } else { + require.NoError(t, err) + require.NotNil(t, conf) + } + }) + } +} + +// newTestWatchInput creates a minimal kubernetesWatchInput for testing +// without requiring a real Kubernetes connection. +func newTestWatchInput() *kubernetesWatchInput { + return &kubernetesWatchInput{ + eventChan: make(chan watchEvent, 100), + shutSig: shutdown.NewSignaller(), + resourceVers: make(map[string]string), + } +} + +func TestCloseTriggersEndOfInput(t *testing.T) { + k := newTestWatchInput() + + // Start a goroutine that will call Read + readDone := make(chan error, 1) + go func() { + ctx := context.Background() + _, _, err := k.Read(ctx) + readDone <- err + }() + + // Give Read time to block on the channel + time.Sleep(10 * time.Millisecond) + + // Close should trigger ErrEndOfInput + ctx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + + err := k.Close(ctx) + require.NoError(t, err) + + // Verify Read returned ErrEndOfInput + select { + case readErr := <-readDone: + assert.ErrorIs(t, readErr, service.ErrEndOfInput) + case <-time.After(time.Second): + t.Fatal("Read did not return after Close") + } +} + +func TestCloseDrainsEventsBeforeShutdown(t *testing.T) { + k := newTestWatchInput() + + // Add some events to the channel before closing + testEvents := []watchEvent{ + {eventType: "ADDED", object: &unstructured.Unstructured{ + Object: map[string]interface{}{ + "apiVersion": "v1", + "kind": "Pod", + "metadata": map[string]interface{}{ + "name": "test-pod-1", + "namespace": "default", + }, + }, + }}, + {eventType: "MODIFIED", object: &unstructured.Unstructured{ + Object: map[string]interface{}{ + "apiVersion": "v1", + "kind": "Pod", + "metadata": map[string]interface{}{ + "name": "test-pod-2", + "namespace": "default", + }, + }, + }}, + } + + for _, evt := range testEvents { + k.eventChan <- evt + } + + // Read the events before closing + ctx := context.Background() + receivedEvents := 0 + + for i := 0; i < len(testEvents); i++ { + msg, ack, err := k.Read(ctx) + require.NoError(t, err) + require.NotNil(t, msg) + require.NotNil(t, ack) + receivedEvents++ + } + + assert.Equal(t, len(testEvents), receivedEvents, "should receive all events before close") + + // Now close + closeCtx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + + err := k.Close(closeCtx) + require.NoError(t, err) + + // Subsequent reads should return ErrEndOfInput + _, _, err = k.Read(ctx) + assert.ErrorIs(t, err, service.ErrEndOfInput) +} + +func TestReadReturnsErrEndOfInputOnClosedChannel(t *testing.T) { + k := newTestWatchInput() + + // Close the channel directly to simulate shutdown + close(k.eventChan) + + ctx := context.Background() + _, _, err := k.Read(ctx) + + assert.ErrorIs(t, err, service.ErrEndOfInput) +} + +func TestReadRespectsContextCancellation(t *testing.T) { + k := newTestWatchInput() + + ctx, cancel := context.WithCancel(context.Background()) + + // Start Read in a goroutine + readDone := make(chan error, 1) + go func() { + _, _, err := k.Read(ctx) + readDone <- err + }() + + // Give Read time to block + time.Sleep(10 * time.Millisecond) + + // Cancel the context + cancel() + + // Verify Read returned context.Canceled + select { + case readErr := <-readDone: + assert.ErrorIs(t, readErr, context.Canceled) + case <-time.After(time.Second): + t.Fatal("Read did not return after context cancellation") + } + + // Clean up + closeCtx, closeCancel := context.WithTimeout(context.Background(), time.Second) + defer closeCancel() + _ = k.Close(closeCtx) +} + +func TestConcurrentReadsAndClose(t *testing.T) { + k := newTestWatchInput() + + // Start multiple readers + const numReaders = 5 + var wg sync.WaitGroup + errors := make(chan error, numReaders) + + for i := 0; i < numReaders; i++ { + wg.Add(1) + go func() { + defer wg.Done() + ctx := context.Background() + _, _, err := k.Read(ctx) + errors <- err + }() + } + + // Give readers time to block + time.Sleep(20 * time.Millisecond) + + // Close the input + closeCtx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + + err := k.Close(closeCtx) + require.NoError(t, err) + + // Wait for all readers to finish + wg.Wait() + close(errors) + + // All readers should have received ErrEndOfInput + for readErr := range errors { + assert.ErrorIs(t, readErr, service.ErrEndOfInput) + } +} diff --git a/internal/impl/kubernetes/package.go b/internal/impl/kubernetes/package.go new file mode 100644 index 0000000000..276009a41a --- /dev/null +++ b/internal/impl/kubernetes/package.go @@ -0,0 +1 @@ +package kubernetes diff --git a/public/components/all/package.go b/public/components/all/package.go index 34fc0f87ab..0ff15b44dd 100644 --- a/public/components/all/package.go +++ b/public/components/all/package.go @@ -31,6 +31,7 @@ import ( _ "github.com/warpstreamlabs/bento/public/components/jaeger" _ "github.com/warpstreamlabs/bento/public/components/javascript" _ "github.com/warpstreamlabs/bento/public/components/kafka" + _ "github.com/warpstreamlabs/bento/public/components/kubernetes" _ "github.com/warpstreamlabs/bento/public/components/maxmind" _ "github.com/warpstreamlabs/bento/public/components/memcached" _ "github.com/warpstreamlabs/bento/public/components/mongodb" diff --git a/public/components/kubernetes/package.go b/public/components/kubernetes/package.go new file mode 100644 index 0000000000..585ed372e6 --- /dev/null +++ b/public/components/kubernetes/package.go @@ -0,0 +1,6 @@ +package kubernetes + +import ( + // Bring in the internal plugin definitions. + _ "github.com/warpstreamlabs/bento/internal/impl/kubernetes" +) diff --git a/website/docs/components/inputs/kubernetes_watch.md b/website/docs/components/inputs/kubernetes_watch.md new file mode 100644 index 0000000000..b1d2e3004e --- /dev/null +++ b/website/docs/components/inputs/kubernetes_watch.md @@ -0,0 +1,426 @@ +--- +title: kubernetes_watch +slug: kubernetes_watch +type: input +status: stable +categories: ["Services","Kubernetes"] +--- + + + +import Tabs from '@theme/Tabs'; +import TabItem from '@theme/TabItem'; + +Watches Kubernetes resources for changes, similar to `kubectl get --watch`. + +Introduced in version 1.15.0. + + + + + + +```yml +# Common config fields, showing default values +input: + label: "" + kubernetes_watch: + auto_auth: true + kubeconfig: "" + kubeconfig_yaml: "" + context: "" + namespaces: [] + label_selector: {} + resource: "" + event_types: + - ADDED + - MODIFIED + - DELETED + include_initial_list: true +``` + + + + +```yml +# All config fields, showing default values +input: + label: "" + kubernetes_watch: + auto_auth: true + kubeconfig: "" + kubeconfig_yaml: "" + context: "" + api_server: "" + token: "" + token_file: "" + ca_file: "" + insecure_skip_verify: false + client_qps: 0 + client_burst: 0 + namespaces: [] + label_selector: {} + field_selector: {} + request_timeout: 30s + resource: "" + custom_resource: + group: "" + version: v1 + resource: "" + event_types: + - ADDED + - MODIFIED + - DELETED + include_initial_list: true + max_retries: 0 + backoff: + initial_interval: 1s + max_interval: 60s + max_elapsed_time: 0s +``` + + + + +This input watches Kubernetes resources and emits events when resources +are added, modified, or deleted. It uses the Kubernetes discovery API +to automatically resolve resource types, supporting both standard +resources and Custom Resource Definitions (CRDs). + +### Watching Resources + +Specify any resource name recognized by the cluster: +```yaml +input: + kubernetes_watch: + resource: pods +``` + +The resource name is resolved dynamically via the Kubernetes API, +so any valid resource type is supported (pods, deployments, services, +configmaps, or any installed CRD). + +### Explicit GVR for Custom Resources + +For custom resources where you need to specify the exact group/version: +```yaml +input: + kubernetes_watch: + custom_resource: + group: stable.example.com + version: v1 + resource: crontabs +``` + +### Watch Events + +Each message includes the watch event type in metadata: +- **ADDED**: Resource was created +- **MODIFIED**: Resource was updated +- **DELETED**: Resource was removed + + +### Metadata + +This input adds the following metadata fields to each message: + +```text +- kubernetes_watch_event_type +- kubernetes_resource_kind +- kubernetes_resource_name +- kubernetes_resource_namespace +- kubernetes_resource_version +- kubernetes_resource_uid +- kubernetes_resource_creation_timestamp +``` + +You can access these metadata fields using +[function interpolation](/docs/configuration/interpolation#bloblang-queries). + + +Additionally, all resource labels are added as metadata with the prefix `kubernetes_labels_`, and all annotations are added with the prefix `kubernetes_annotations_`. For example, a label `app: myapp` becomes metadata key `kubernetes_labels_app`. + + +## Fields + +### `auto_auth` + +Automatically detect authentication method. Tries in-cluster config first, then falls back to kubeconfig. + + +Type: `bool` +Default: `true` + +### `kubeconfig` + +Path to kubeconfig file. If empty, uses $KUBECONFIG (if set) or the default kubeconfig location (~/.kube/config). + + +Type: `string` +Default: `""` + +### `kubeconfig_yaml` + +Kubeconfig content as a YAML string. Use this to embed kubeconfig directly in your config (e.g., from a secret or environment variable) instead of referencing a file path. +:::warning Secret +This field contains sensitive information that usually shouldn't be added to a config directly, read our [secrets page for more info](/docs/configuration/secrets). +::: + + +Type: `string` +Default: `""` + +### `context` + +Kubernetes context to use from kubeconfig. If empty, uses the current context. + + +Type: `string` +Default: `""` + +### `api_server` + +Kubernetes API server URL. Only used when providing explicit credentials. + + +Type: `string` +Default: `""` + +### `token` + +Bearer token for authentication. Can be a service account token. +:::warning Secret +This field contains sensitive information that usually shouldn't be added to a config directly, read our [secrets page for more info](/docs/configuration/secrets). +::: + + +Type: `string` +Default: `""` + +### `token_file` + +Path to file containing bearer token. + + +Type: `string` +Default: `""` + +### `ca_file` + +Path to CA certificate file for verifying API server. + + +Type: `string` +Default: `""` + +### `insecure_skip_verify` + +Skip TLS certificate verification. Not recommended for production. + + +Type: `bool` +Default: `false` + +### `client_qps` + +QPS limit for Kubernetes API client. 0 uses the client-go default. + + +Type: `float` +Default: `0` + +### `client_burst` + +Burst limit for Kubernetes API client. 0 uses the client-go default. + + +Type: `int` +Default: `0` + +### `namespaces` + +Namespaces to watch. Empty list means all namespaces. + + +Type: `array` +Default: `[]` + +```yml +# Examples + +namespaces: + - default + +namespaces: + - production + - staging +``` + +### `label_selector` + +Kubernetes label selector to filter resources. + + +Type: `object` +Default: `{}` + +```yml +# Examples + +label_selector: + app: myapp + +label_selector: + app: myapp + env: prod +``` + +### `field_selector` + +Kubernetes field selector to filter resources. + + +Type: `object` +Default: `{}` + +```yml +# Examples + +field_selector: + status.phase: Running + +field_selector: + metadata.name: my-pod +``` + +### `request_timeout` + +Timeout for Kubernetes API requests such as list calls. Use "0s" to disable. + + +Type: `string` +Default: `"30s"` + +### `resource` + +Standard Kubernetes resource type to watch. + + +Type: `string` +Default: `""` + +```yml +# Examples + +resource: pods + +resource: deployments + +resource: configmaps +``` + +### `custom_resource` + +Custom Resource Definition to watch. Use this for CRDs instead of 'resource'. + + +Type: `object` + +### `custom_resource.group` + +API group for the custom resource (e.g., 'stable.example.com'). + + +Type: `string` +Default: `""` + +### `custom_resource.version` + +API version for the custom resource (e.g., 'v1', 'v1beta1'). + + +Type: `string` +Default: `"v1"` + +### `custom_resource.resource` + +Plural name of the custom resource (e.g., 'crontabs'). + + +Type: `string` +Default: `""` + +### `event_types` + +Watch event types to include. Valid values are `ADDED`, `MODIFIED`, and `DELETED`. + + +Type: `array` +Default: `["ADDED","MODIFIED","DELETED"]` + +```yml +# Examples + +event_types: + - ADDED + - DELETED + +event_types: + - MODIFIED +``` + +### `include_initial_list` + +Emit ADDED events for all existing resources when starting. + + +Type: `bool` +Default: `true` + +### `max_retries` + +The maximum number of retries before giving up on the request. If set to zero there is no discrete limit. + + +Type: `int` +Default: `0` + +### `backoff` + +Control time intervals between retry attempts. + + +Type: `object` + +### `backoff.initial_interval` + +The initial period to wait between retry attempts. + + +Type: `string` +Default: `"1s"` + +### `backoff.max_interval` + +The maximum period to wait between retry attempts. + + +Type: `string` +Default: `"60s"` + +### `backoff.max_elapsed_time` + +The maximum period to wait before retry attempts are abandoned. If zero then no limit is used. + + +Type: `string` +Default: `"0s"` + +