diff --git a/README.md b/README.md index cb8a3e2..7754ab3 100644 --- a/README.md +++ b/README.md @@ -4,3 +4,4 @@ Reference architectures for various use cases at CoreWeave. ## Additional Reference Areas - [secrets-management](./secrets-management/README.md): Cloud KMS-backed secret manager patterns for Kubernetes workloads. +- [kafka-local-nvme](./kafka-local-nvme/README.md): Strimzi Kafka reference for CoreWeave local NVMe, with opt-in local-PV repair. diff --git a/kafka-local-nvme/.dockerignore b/kafka-local-nvme/.dockerignore new file mode 100644 index 0000000..e9dcab0 --- /dev/null +++ b/kafka-local-nvme/.dockerignore @@ -0,0 +1,2 @@ +.git +tests diff --git a/kafka-local-nvme/Dockerfile b/kafka-local-nvme/Dockerfile new file mode 100644 index 0000000..8af61c0 --- /dev/null +++ b/kafka-local-nvme/Dockerfile @@ -0,0 +1,12 @@ +FROM golang:1.24-alpine AS build +WORKDIR /src +COPY go.mod ./ +RUN go mod download +COPY cmd ./cmd +COPY internal ./internal +RUN CGO_ENABLED=0 go test ./... +RUN CGO_ENABLED=0 go build -trimpath -ldflags='-s -w' -o /out/kafka-local-pv-repair ./cmd/kafka-local-pv-repair + +FROM gcr.io/distroless/static-debian12:nonroot +COPY --from=build /out/kafka-local-pv-repair /kafka-local-pv-repair +ENTRYPOINT ["/kafka-local-pv-repair"] diff --git a/kafka-local-nvme/LICENSE b/kafka-local-nvme/LICENSE new file mode 100644 index 0000000..dc8b82a --- /dev/null +++ b/kafka-local-nvme/LICENSE @@ -0,0 +1,203 @@ +Copyright 2026 Chris Milsted + + Apache License + Version 2.0, January 2004 + http://www.apache.org/licenses/ + + TERMS AND CONDITIONS FOR USE, REPRODUCTION, AND DISTRIBUTION + + 1. Definitions. + + "License" shall mean the terms and conditions for use, reproduction, + and distribution as defined by Sections 1 through 9 of this document. + + "Licensor" shall mean the copyright owner or entity authorized by + the copyright owner that is granting the License. + + "Legal Entity" shall mean the union of the acting entity and all + other entities that control, are controlled by, or are under common + control with that entity. For the purposes of this definition, + "control" means (i) the power, direct or indirect, to cause the + direction or management of such entity, whether by contract or + otherwise, or (ii) ownership of fifty percent (50%) or more of the + outstanding shares, or (iii) beneficial ownership of such entity. + + "You" (or "Your") shall mean an individual or Legal Entity + exercising permissions granted by this License. + + "Source" form shall mean the preferred form for making modifications, + including but not limited to software source code, documentation + source, and configuration files. + + "Object" form shall mean any form resulting from mechanical + transformation or translation of a Source form, including but + not limited to compiled object code, generated documentation, + and conversions to other media types. + + "Work" shall mean the work of authorship, whether in Source or + Object form, made available under the License, as indicated by a + copyright notice that is included in or attached to the work + (an example is provided in the Appendix below). + + "Derivative Works" shall mean any work, whether in Source or Object + form, that is based on (or derived from) the Work and for which the + editorial revisions, annotations, elaborations, or other modifications + represent, as a whole, an original work of authorship. For the purposes + of this License, Derivative Works shall not include works that remain + separable from, or merely link (or bind by name) to the interfaces of, + the Work and Derivative Works thereof. + + "Contribution" shall mean any work of authorship, including + the original version of the Work and any modifications or additions + to that Work or Derivative Works thereof, that is intentionally + submitted to Licensor for inclusion in the Work by the copyright owner + or by an individual or Legal Entity authorized to submit on behalf of + the copyright owner. For the purposes of this definition, "submitted" + means any form of electronic, verbal, or written communication sent + to the Licensor or its representatives, including but not limited to + communication on electronic mailing lists, source code control systems, + and issue tracking systems that are managed by, or on behalf of, the + Licensor for the purpose of discussing and improving the Work, but + excluding communication that is conspicuously marked or otherwise + designated in writing by the copyright owner as "Not a Contribution." + + "Contributor" shall mean Licensor and any individual or Legal Entity + on behalf of whom a Contribution has been received by Licensor and + subsequently incorporated within the Work. + + 2. Grant of Copyright License. Subject to the terms and conditions of + this License, each Contributor hereby grants to You a perpetual, + worldwide, non-exclusive, no-charge, royalty-free, irrevocable + copyright license to reproduce, prepare Derivative Works of, + publicly display, publicly perform, sublicense, and distribute the + Work and such Derivative Works in Source or Object form. + + 3. Grant of Patent License. Subject to the terms and conditions of + this License, each Contributor hereby grants to You a perpetual, + worldwide, non-exclusive, no-charge, royalty-free, irrevocable + (except as stated in this section) patent license to make, have made, + use, offer to sell, sell, import, and otherwise transfer the Work, + where such license applies only to those patent claims licensable + by such Contributor that are necessarily infringed by their + Contribution(s) alone or by combination of their Contribution(s) + with the Work to which such Contribution(s) was submitted. If You + institute patent litigation against any entity (including a + cross-claim or counterclaim in a lawsuit) alleging that the Work + or a Contribution incorporated within the Work constitutes direct + or contributory patent infringement, then any patent licenses + granted to You under this License for that Work shall terminate + as of the date such litigation is filed. + + 4. Redistribution. You may reproduce and distribute copies of the + Work or Derivative Works thereof in any medium, with or without + modifications, and in Source or Object form, provided that You + meet the following conditions: + + (a) You must give any other recipients of the Work or + Derivative Works a copy of this License; and + + (b) You must cause any modified files to carry prominent notices + stating that You changed the files; and + + (c) You must retain, in the Source form of any Derivative Works + that You distribute, all copyright, patent, trademark, and + attribution notices from the Source form of the Work, + excluding those notices that do not pertain to any part of + the Derivative Works; and + + (d) If the Work includes a "NOTICE" text file as part of its + distribution, then any Derivative Works that You distribute must + include a readable copy of the attribution notices contained + within such NOTICE file, excluding those notices that do not + pertain to any part of the Derivative Works, in at least one + of the following places: within a NOTICE text file distributed + as part of the Derivative Works; within the Source form or + documentation, if provided along with the Derivative Works; or, + within a display generated by the Derivative Works, if and + wherever such third-party notices normally appear. The contents + of the NOTICE file are for informational purposes only and + do not modify the License. You may add Your own attribution + notices within Derivative Works that You distribute, alongside + or as an addendum to the NOTICE text from the Work, provided + that such additional attribution notices cannot be construed + as modifying the License. + + You may add Your own copyright statement to Your modifications and + may provide additional or different license terms and conditions + for use, reproduction, or distribution of Your modifications, or + for any such Derivative Works as a whole, provided Your use, + reproduction, and distribution of the Work otherwise complies with + the conditions stated in this License. + + 5. Submission of Contributions. Unless You explicitly state otherwise, + any Contribution intentionally submitted for inclusion in the Work + by You to the Licensor shall be under the terms and conditions of + this License, without any additional terms or conditions. + Notwithstanding the above, nothing herein shall supersede or modify + the terms of any separate license agreement you may have executed + with Licensor regarding such Contributions. + + 6. Trademarks. This License does not grant permission to use the trade + names, trademarks, service marks, or product names of the Licensor, + except as required for reasonable and customary use in describing the + origin of the Work and reproducing the content of the NOTICE file. + + 7. Disclaimer of Warranty. Unless required by applicable law or + agreed to in writing, Licensor provides the Work (and each + Contributor provides its Contributions) on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or + implied, including, without limitation, any warranties or conditions + of TITLE, NON-INFRINGEMENT, MERCHANTABILITY, or FITNESS FOR A + PARTICULAR PURPOSE. You are solely responsible for determining the + appropriateness of using or redistributing the Work and assume any + risks associated with Your exercise of permissions under this License. + + 8. Limitation of Liability. In no event and under no legal theory, + whether in tort (including negligence), contract, or otherwise, + unless required by applicable law (such as deliberate and grossly + negligent acts) or agreed to in writing, shall any Contributor be + liable to You for damages, including any direct, indirect, special, + incidental, or consequential damages of any character arising as a + result of this License or out of the use or inability to use the + Work (including but not limited to damages for loss of goodwill, + work stoppage, computer failure or malfunction, or any and all + other commercial damages or losses), even if such Contributor + has been advised of the possibility of such damages. + + 9. Accepting Warranty or Additional Liability. While redistributing + the Work or Derivative Works thereof, You may choose to offer, + and charge a fee for, acceptance of support, warranty, indemnity, + or other liability obligations and/or rights consistent with this + License. However, in accepting such obligations, You may act only + on Your own behalf and on Your sole responsibility, not on behalf + of any other Contributor, and only if You agree to indemnify, + defend, and hold each Contributor harmless for any liability + incurred by, or claims asserted against, such Contributor by reason + of your accepting any such warranty or additional liability. + + END OF TERMS AND CONDITIONS + + APPENDIX: How to apply the Apache License to your work. + + To apply the Apache License to your work, attach the following + boilerplate notice, with the fields enclosed by brackets "[]" + replaced with your own identifying information. (Don't include + the brackets!) The text should be enclosed in the appropriate + comment syntax for the file format. We also recommend that a + file or class name and description of purpose be included on the + same "printed page" as the copyright notice for easier + identification within third-party archives. + + Copyright [yyyy] [name of copyright owner] + + Licensed under the Apache License, Version 2.0 (the "License"); + you may not use this file except in compliance with the License. + You may obtain a copy of the License at + + http://www.apache.org/licenses/LICENSE-2.0 + + Unless required by applicable law or agreed to in writing, software + distributed under the License is distributed on an "AS IS" BASIS, + WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + See the License for the specific language governing permissions and + limitations under the License. diff --git a/kafka-local-nvme/README.md b/kafka-local-nvme/README.md new file mode 100644 index 0000000..c0a27a8 --- /dev/null +++ b/kafka-local-nvme/README.md @@ -0,0 +1,111 @@ +# Kafka on CoreWeave local NVMe + + +[Customer reference architecture](docs/customer-reference-architecture.md) + +This non-production reference deploys Strimzi Kafka with dynamically provisioned, node-local storage below `/mnt/local/kafka`. It is a deployment starting point, not a production architecture, recovery procedure, or validation of CoreWeave availability, performance, or security. + +## Important storage limits + +`/mnt/local` is node-local storage. Treat its contents as lost when a node is lost or rebooted; Kubernetes PV/PVC identity does not make the underlying bytes durable. + +The `kafka-local` StorageClass uses `Retain` and the node pools use `deleteClaim: false`. Removing Kafka can therefore leave PVs, PVCs, and local data behind. Reusing the same namespace and Kafka resource names can reuse the same PVC-named directories; plan cleanup and identity changes deliberately. + +The provisioner creates directories and local PVs by running a Kubernetes helper Pod on the scheduled node. `WaitForFirstConsumer` ensures that node is selected first. Directory-backed local-path volumes can exceed their PVC request: requests neither reserve nor enforce `/mnt/local` capacity. Operators must independently verify available capacity and leave operational headroom on every selected node. + +## Profiles + +| Profile | Brokers | Controllers | Broker storage | Topic RF / min ISR | +|---|---:|---:|---:|---:| +| `two-node` | 2 | 3 | 400Gi each | 2 / 1 | +| `five-node` | 5 | 3 | 400Gi each | 5 / 3 | + +Profiles are mutually exclusive and use the same resource names. The three controllers each request 20Gi. +The five-node profile also enables Cruise Control and uses larger CPU and memory requests than the two-node profile. + +## Prerequisites + +- A Kubernetes cluster on CoreWeave, with `/mnt/local` available on every selected node. +- `kubectl`, cluster-admin access for the local-path provisioner, and a Strimzi Cluster Operator configured to watch and reconcile the `kafka` namespace. +- At least two or five nodes, respectively, labelled `kafka.local/kafka-local=true`. +- Enough free node-local capacity for the requested claims and Kafka operational headroom. + +Install the Strimzi operator using its [official installation instructions](https://strimzi.io/docs/operators/latest/deploying.html) before applying a profile. + +## Deploy + +From this directory: + +```bash +kubectl apply -k provisioner/local-path +kubectl create namespace kafka --dry-run=client -o yaml | kubectl apply -f - +kubectl label node NODE_NAME kafka.local/kafka-local=true +kubectl apply -k profiles/two-node +``` + +Use `five-node` in the last command for the five-broker profile. + +Wait for the operator and Kafka resources to reconcile: + +```bash +kubectl wait -n kafka --for=condition=Ready kafka/kafka-local --timeout=10m +kubectl wait -n kafka --for=condition=Ready kafkatopic/kafka-local-topic --timeout=10m +kubectl get kafka,kafkanodepool,kafkatopic,pvc -n kafka +kubectl get pods -n kafka -o wide +``` + +Confirm that each PV is local and bound to the node chosen for its consuming Kafka Pod: + +```bash +kubectl get pv -o custom-columns=NAME:.metadata.name,PATH:.spec.local.path,NODE:.spec.nodeAffinity.required.nodeSelectorTerms[0].matchExpressions[0].values[0],CLAIM_NAMESPACE:.spec.claimRef.namespace,CLAIM_NAME:.spec.claimRef.name +``` + +After Kafka and `kafka-local-topic` are Ready, run the basic producer/readback check: + +```bash +./scripts/smoke-test.sh +``` + +It verifies exact readback of a small uniquely tagged message set produced with `acks=all`; it does not test node loss, reboot recovery, HA, performance, or security. + +## Optional Repair DaemonSet + +The normal Kustomizations do not install repair. The optional Repair DaemonSet only recreates absent directories in the exact managed hierarchy `/mnt/local/kafka//` after it verifies the host mount and an authorized, unchanged Node identity. It never restores lost bytes, deletes anything, or mutates PVs or PVCs. Kafka can rebuild only from healthy replicas; this is not automatic reboot, replacement, cleanup, or provider-behavior proof. + +Use it only during serialized operator maintenance: + +1. Build and publish the image, then use its immutable `repository@sha256:...` digest. The image build runs the Linux test suite before compiling the binary. + + ```bash + docker build -t registry.example/kafka-local-pv-repair:VERSION . + docker push registry.example/kafka-local-pv-repair:VERSION + ``` +2. Capture each Node's exact `name`, hostname, Kubernetes UID, and nonempty provider ID in a tab-separated allowlist. Any identity difference requires explicit reauthorization and a newly rendered manifest. +3. Independently verify the host `/mnt/local` mount filesystem, source, and canonical sorted comma-separated mount options; omit the `ro`/`rw` mode because the agent verifies that the host mount is writable itself. +4. Render and apply the complete generated manifest (including ServiceAccount and RBAC). The renderer requires Python 3. + + ```bash + ./scripts/render-kafka-local-pv-repair-manifest.sh \ + --image registry.example/kafka-local-pv-repair@sha256:REPLACE_WITH_64_HEX \ + --mount-fs ext4 --mount-source /dev/REPLACE_ME \ + --mount-options nodev,nosuid \ + --nodes nodes.tsv | kubectl apply -f - + ``` + +5. Obtain the required Pod Security approval for `hostPID` and a read-write `/mnt/local` hostPath. The container runs as root only to create missing root-owned hierarchy components, with `CHOWN` and `DAC_OVERRIDE` as its only added capabilities. Check the DaemonSet rollout and logs, and serialize all related operator maintenance. A newly rendered authorization changes the Pod template and rolls the DaemonSet automatically. + +The two-node profile can lose controller quorum. Simultaneous local-storage loss can be unrecoverable even with the five-node profile. + +## Validate manifests + +```bash +./scripts/validate.sh +kubectl apply --dry-run=server -k provisioner/local-path +kubectl apply --dry-run=server -k profiles/two-node +``` + +The server dry run is optional and requires a cluster with the Strimzi CRDs installed. Run it again with `profiles/five-node` when using that profile. + +## Package license + +This package is governed by [LICENSE](LICENSE) (Apache-2.0) and [THIRD_PARTY_NOTICES.md](THIRD_PARTY_NOTICES.md). diff --git a/kafka-local-nvme/THIRD_PARTY_NOTICES.md b/kafka-local-nvme/THIRD_PARTY_NOTICES.md new file mode 100644 index 0000000..489e3d1 --- /dev/null +++ b/kafka-local-nvme/THIRD_PARTY_NOTICES.md @@ -0,0 +1,13 @@ +# Third-party notices + +## Rancher Local Path Provisioner + +The manifests under `provisioner/local-path/` are derived from Rancher Local +Path Provisioner version 0.0.36: + + + +Copyright 2014-2020 Rancher Labs, Inc. + +Rancher Local Path Provisioner is licensed under the Apache License, Version +2.0. A copy of that license is included in this package as `LICENSE`. diff --git a/kafka-local-nvme/cmd/kafka-local-pv-repair/main.go b/kafka-local-nvme/cmd/kafka-local-pv-repair/main.go new file mode 100644 index 0000000..c584b04 --- /dev/null +++ b/kafka-local-nvme/cmd/kafka-local-pv-repair/main.go @@ -0,0 +1,40 @@ +//go:build linux + +package main + +import ( + "context" + "log" + "os" + "time" + + "github.com/coreweave/reference-architecture/kafka-local-nvme/internal/repair" +) + +func main() { + config, err := repair.LoadConfig(env("REPAIR_CONFIG_PATH", "/etc/repair/config.json")) + if err != nil { + log.Fatalf("invalid repair config: %v", err) + } + reader, err := repair.NewInClusterReader() + if err != nil { + log.Fatalf("Kubernetes API initialization failed: %v", err) + } + allowed := repair.Allowlist{} + for _, node := range config.Nodes { + allowed[node.Name] = node + } + engine := &repair.Engine{Reader: reader, Allowlist: allowed, NodeName: os.Getenv("NODE_NAME"), MountInfo: "/proc/1/mountinfo", Mount: repair.MountSignature{Source: config.MountSource, FSType: config.MountFilesystem, Options: config.MountOptions}} + for { + if err := engine.Repair(context.Background()); err != nil { + log.Printf("repair reconciliation refused: %v", err) + } + time.Sleep(30 * time.Second) + } +} +func env(key, fallback string) string { + if value := os.Getenv(key); value != "" { + return value + } + return fallback +} diff --git a/kafka-local-nvme/docs/customer-reference-architecture.md b/kafka-local-nvme/docs/customer-reference-architecture.md new file mode 100644 index 0000000..89b59ed --- /dev/null +++ b/kafka-local-nvme/docs/customer-reference-architecture.md @@ -0,0 +1,48 @@ +# Kafka on CoreWeave local NVMe: customer reference + +## Scope and status + + + +This is a non-production deployment reference for Strimzi Kafka using CoreWeave node-local storage. It supplies a local-path provisioner configuration and two mutually exclusive Kafka profiles. It does not claim recovery from node loss or reboot, CoreWeave validation, HA, performance, security, compliance, or production readiness. + +## What it deploys + +The local-path provisioner creates a helper Pod after scheduling selects a node. The helper writes a PVC-named directory under `/mnt/local/kafka`, and the provisioner creates a local PV with node affinity. `WaitForFirstConsumer` prevents provisioning before that node is selected. + +| Profile | Brokers | Controllers | Topic RF / min ISR | +|---|---:|---:|---:| +| `two-node` | 2 | 3 | 2 / 1 | +| `five-node` | 5 | 3 | 5 / 3 | + +Both profiles select nodes labelled `kafka.local/kafka-local=true`, use hostname placement, and create 400Gi broker claims plus 20Gi controller claims. They share names and cannot be installed together. +The five-node profile enables Cruise Control and uses larger CPU and memory requests than the two-node profile. + +## Storage lifecycle + +`/mnt/local` is node-local and volatile: loss or reboot of a node can make its data unavailable. A local PV and its PVC keep Kubernetes identity during ordinary Pod replacement on the same running node; they do not preserve data through node loss or reboot. + +The StorageClass has `Retain`, and Strimzi node pools set `deleteClaim: false`. Deleting Kafka can leave PVs, PVCs, and local directories. A later deployment with the same namespace and PVC names may encounter or reuse those directories. Establish an operations and data-retention process before use. + +PVC sizes are not hard limits for directory-backed local-path volumes: directory-backed local-path PVC requests neither reserve nor enforce `/mnt/local` capacity. Operators must independently verify available capacity and leave operational headroom on every selected node. + +## Deployment and verification + +Install Strimzi, apply `provisioner/local-path`, label the intended nodes, and apply one profile. See the [package README](../README.md) for commands. + +Verify that PVCs bind, Kafka resources become Ready, and every PV reports a local path and node affinity consistent with its Kafka Pod. The included [smoke test](../scripts/smoke-test.sh) produces with `acks=all` and verifies exact readback of a uniquely tagged message set. It is a basic deployment check only. + +## Optional repair boundary + +The repository does not install the Repair DaemonSet through its normal Kustomizations. When explicitly rendered and applied, it is a privileged host-storage component: it uses `hostPID` to inspect `/proc/1/mountinfo` and a read-write hostPath for `/mnt/local`. Its trust boundary requires a Pod Security exception, deliberate operator ownership, and serialized maintenance. + +The agent may recreate only missing components of the exact managed directory hierarchy for a Bound local PV with an authorized same Node identity and a verified host mount signature. It does not restore data, delete directories, or write Kubernetes PV/PVC objects. Kafka rebuilds only from healthy replicas; this is not a claim of automatic reboot or replacement recovery, cleanup, or provider validation. It runs as root solely to create missing root-owned directories, with only `CHOWN` and `DAC_OVERRIDE` added after dropping all capabilities. + +Operators build and publish the image themselves, then render with its immutable digest. The Docker build runs Linux tests. Rendering requires Python 3, a captured exact Node name/hostname/UID/provider-ID allowlist, and independently verified mount filesystem, source, and canonical sorted options excluding `ro`/`rw`. Any identity mismatch requires reauthorization and a newly rendered ConfigMap; the changed Pod template rolls the DaemonSet automatically. Check rollout status and logs, and keep maintenance serialized. The two-node profile can lose controller quorum; simultaneous loss can be unrecoverable. + +## Customer decisions before deployment + +- Whether node-local loss is acceptable and what replication, backup, and recovery design is required. +- Node selection, capacity reservation, quotas, monitoring, and operational headroom. +- Secure listener exposure, network policy, authentication, authorization, encryption, observability, and lifecycle ownership. +- A controlled retained-storage and decommissioning process that avoids unintended same-name reuse. diff --git a/kafka-local-nvme/go.mod b/kafka-local-nvme/go.mod new file mode 100644 index 0000000..0afea65 --- /dev/null +++ b/kafka-local-nvme/go.mod @@ -0,0 +1,3 @@ +module github.com/coreweave/reference-architecture/kafka-local-nvme + +go 1.24.0 diff --git a/kafka-local-nvme/internal/repair/api.go b/kafka-local-nvme/internal/repair/api.go new file mode 100644 index 0000000..3974cd0 --- /dev/null +++ b/kafka-local-nvme/internal/repair/api.go @@ -0,0 +1,163 @@ +package repair + +import ( + "context" + "crypto/tls" + "crypto/x509" + "encoding/json" + "errors" + "fmt" + "net/http" + "net/url" + "os" + "path/filepath" +) + +var ErrUnavailable = errors.New("Kubernetes reader is not configured") + +type APIReader struct { + client *http.Client + base, token string +} + +func NewInClusterReader() (*APIReader, error) { + token, err := os.ReadFile("/var/run/secrets/kubernetes.io/serviceaccount/token") + if err != nil { + return nil, err + } + ca, err := os.ReadFile("/var/run/secrets/kubernetes.io/serviceaccount/ca.crt") + if err != nil { + return nil, err + } + pool := x509.NewCertPool() + if !pool.AppendCertsFromPEM(ca) { + return nil, errors.New("invalid service account CA") + } + host, port := os.Getenv("KUBERNETES_SERVICE_HOST"), os.Getenv("KUBERNETES_SERVICE_PORT") + if host == "" || port == "" { + return nil, ErrUnavailable + } + return &APIReader{&http.Client{Transport: &http.Transport{TLSClientConfig: &tls.Config{RootCAs: pool, MinVersion: tls.VersionTLS12}}}, "https://" + host + ":" + port, string(token)}, nil +} +func (r *APIReader) get(ctx context.Context, path string, out any) error { + req, err := http.NewRequestWithContext(ctx, http.MethodGet, r.base+path, nil) + if err != nil { + return err + } + req.Header.Set("Authorization", "Bearer "+r.token) + res, err := r.client.Do(req) + if err != nil { + return err + } + defer res.Body.Close() + if res.StatusCode != http.StatusOK { + return fmt.Errorf("Kubernetes GET %s: %s", path, res.Status) + } + return json.NewDecoder(res.Body).Decode(out) +} + +type object struct { + Metadata struct { + Name string `json:"name"` + Namespace string `json:"namespace"` + UID string `json:"uid"` + } `json:"metadata"` + Status struct { + Phase string `json:"phase"` + } `json:"status"` + Spec struct { + StorageClassName string `json:"storageClassName"` + ClaimRef *struct { + Namespace string `json:"namespace"` + Name string `json:"name"` + UID string `json:"uid"` + } `json:"claimRef"` + Local *struct { + Path string `json:"path"` + } `json:"local"` + NodeAffinity *struct { + Required struct { + NodeSelectorTerms []struct { + MatchExpressions []struct { + Key string `json:"key"` + Operator string `json:"operator"` + Values []string `json:"values"` + } `json:"matchExpressions"` + } `json:"nodeSelectorTerms"` + } `json:"required"` + } `json:"nodeAffinity"` + VolumeName string `json:"volumeName"` + } `json:"spec"` +} + +func pv(o object) PV { + p := PV{Name: o.Metadata.Name, Phase: o.Status.Phase, VolumeMode: "local", StorageClass: o.Spec.StorageClassName} + if o.Spec.ClaimRef != nil { + p.ClaimNamespace = o.Spec.ClaimRef.Namespace + p.ClaimName = o.Spec.ClaimRef.Name + p.ClaimUID = o.Spec.ClaimRef.UID + } + if o.Spec.Local != nil { + p.LocalPath = o.Spec.Local.Path + } + if o.Spec.NodeAffinity != nil && len(o.Spec.NodeAffinity.Required.NodeSelectorTerms) == 1 && len(o.Spec.NodeAffinity.Required.NodeSelectorTerms[0].MatchExpressions) == 1 { + e := o.Spec.NodeAffinity.Required.NodeSelectorTerms[0].MatchExpressions[0] + if e.Key == "kubernetes.io/hostname" && e.Operator == "In" && len(e.Values) == 1 { + p.NodeHostname = e.Values[0] + p.NodeOperator = e.Operator + p.NodeAffinityValid = true + } + } + return p +} +func (r *APIReader) ListPVs(c context.Context) ([]PV, error) { + var x struct { + Items []object `json:"items"` + } + if err := r.get(c, "/api/v1/persistentvolumes", &x); err != nil { + return nil, err + } + out := make([]PV, 0, len(x.Items)) + for _, o := range x.Items { + out = append(out, pv(o)) + } + return out, nil +} +func (r *APIReader) GetPV(c context.Context, n string) (PV, error) { + var x object + err := r.get(c, "/api/v1/persistentvolumes/"+url.PathEscape(n), &x) + return pv(x), err +} +func (r *APIReader) GetPVC(c context.Context, ns, n string) (PVC, error) { + var x object + err := r.get(c, "/api/v1/namespaces/"+url.PathEscape(ns)+"/persistentvolumeclaims/"+url.PathEscape(n), &x) + return PVC{Namespace: x.Metadata.Namespace, Name: x.Metadata.Name, UID: x.Metadata.UID, VolumeName: x.Spec.VolumeName, Phase: x.Status.Phase}, err +} +func (r *APIReader) GetNode(c context.Context, n string) (NodeIdentity, error) { + var x struct { + Metadata struct { + Name string `json:"name"` + UID string `json:"uid"` + Labels map[string]string `json:"labels"` + } `json:"metadata"` + Spec struct { + ProviderID string `json:"providerID"` + } `json:"spec"` + } + err := r.get(c, "/api/v1/nodes/"+url.PathEscape(n), &x) + return NodeIdentity{Name: x.Metadata.Name, Hostname: x.Metadata.Labels["kubernetes.io/hostname"], UID: x.Metadata.UID, ProviderID: x.Spec.ProviderID}, err +} +func LoadConfig(path string) (Config, error) { + b, err := os.ReadFile(filepath.Clean(path)) + if err != nil { + return Config{}, err + } + var c Config + if err = json.Unmarshal(b, &c); err != nil { + return Config{}, err + } + if c.MountFilesystem == "" || c.MountSource == "" || len(c.MountOptions) == 0 || len(c.Nodes) == 0 { + return Config{}, errors.New("incomplete repair config") + } + return c, nil +} diff --git a/kafka-local-nvme/internal/repair/engine_linux.go b/kafka-local-nvme/internal/repair/engine_linux.go new file mode 100644 index 0000000..216e95b --- /dev/null +++ b/kafka-local-nvme/internal/repair/engine_linux.go @@ -0,0 +1,51 @@ +//go:build linux + +package repair + +import ( + "context" + "fmt" +) + +type Engine struct { + Reader Reader + Allowlist Allowlist + NodeName string + MountInfo string + Mount MountSignature +} + +func (e *Engine) Repair(ctx context.Context) error { + candidates, err := Candidates(ctx, e.Reader, e.Allowlist, e.NodeName) + if err != nil { + return err + } + if err = VerifyHostMount(e.MountInfo, e.Mount); err != nil { + return err + } + for _, candidate := range candidates { + pv, err := e.Reader.GetPV(ctx, candidate.PV.Name) + if err != nil { + return err + } + pvc, err := e.Reader.GetPVC(ctx, candidate.PVC.Namespace, candidate.PVC.Name) + if err != nil { + return err + } + node, err := e.Reader.GetNode(ctx, candidate.Node.Name) + if err != nil { + return err + } + expected, _ := ExpectedPath(pv.ClaimNamespace, pv.ClaimName) + if pv != candidate.PV || pvc != candidate.PVC || node != candidate.Node || node.Name != e.NodeName || pv.LocalPath != expected || VerifyIdentity(e.Allowlist, node) != nil { + return fmt.Errorf("candidate %s changed before mutation", candidate.PV.Name) + } + if err = VerifyHostMount(e.MountInfo, e.Mount); err != nil { + return err + } + if err = EnsureHierarchy(expected); err != nil { + return err + } + } + return nil +} diff --git a/kafka-local-nvme/internal/repair/fs_linux.go b/kafka-local-nvme/internal/repair/fs_linux.go new file mode 100644 index 0000000..0193b2a --- /dev/null +++ b/kafka-local-nvme/internal/repair/fs_linux.go @@ -0,0 +1,78 @@ +//go:build linux + +package repair + +import ( + "fmt" + "strings" + "syscall" +) + +// EnsureHierarchy creates only absent directory components under /mnt/local, anchored by FDs. +func EnsureHierarchy(expected string) error { + return ensureHierarchy(HostRoot, expected) +} + +func ensureHierarchy(hostRoot, expected string) error { + prefix := hostRoot + "/" + if !strings.HasPrefix(expected, prefix) || strings.TrimPrefix(expected, prefix) == "" { + return fmt.Errorf("path escapes host root") + } + root, err := syscall.Open(hostRoot, syscall.O_RDONLY|syscall.O_DIRECTORY|syscall.O_NOFOLLOW|syscall.O_CLOEXEC, 0) + if err != nil { + return err + } + defer syscall.Close(root) + var rootStat syscall.Stat_t + if err = syscall.Fstat(root, &rootStat); err != nil { + return err + } + fd := root + for _, part := range strings.Split(strings.TrimPrefix(expected, prefix), "/") { + if part == "" || part == "." || part == ".." { + return fmt.Errorf("unsafe path component") + } + next, openErr := syscall.Openat(fd, part, syscall.O_RDONLY|syscall.O_DIRECTORY|syscall.O_NOFOLLOW|syscall.O_CLOEXEC, 0) + created := false + if openErr == syscall.ENOENT { + if err = syscall.Mkdirat(fd, part, 0770); err != nil { + if err != syscall.EEXIST { + return err + } + } else { + created = true + } + next, openErr = syscall.Openat(fd, part, syscall.O_RDONLY|syscall.O_DIRECTORY|syscall.O_NOFOLLOW|syscall.O_CLOEXEC, 0) + } + if openErr != nil { + return openErr + } + var st syscall.Stat_t + if err = syscall.Fstat(next, &st); err != nil { + syscall.Close(next) + return err + } + if st.Dev != rootStat.Dev { + syscall.Close(next) + return fmt.Errorf("cross-mount component %q", part) + } + if created { + if err = syscall.Fchmod(next, 0770); err != nil { + syscall.Close(next) + return err + } + if err = syscall.Fchown(next, 1001, 0); err != nil { + syscall.Close(next) + return err + } + } + if fd != root { + syscall.Close(fd) + } + fd = next + } + if fd != root { + defer syscall.Close(fd) + } + return nil +} diff --git a/kafka-local-nvme/internal/repair/identity.go b/kafka-local-nvme/internal/repair/identity.go new file mode 100644 index 0000000..9ac62b6 --- /dev/null +++ b/kafka-local-nvme/internal/repair/identity.go @@ -0,0 +1,17 @@ +package repair + +import "fmt" + +func VerifyIdentity(allowed Allowlist, live NodeIdentity) error { + want, ok := allowed[live.Name] + if !ok { + return fmt.Errorf("node %q is not allowlisted", live.Name) + } + if live.Name == "" || live.Hostname == "" || live.UID == "" || live.ProviderID == "" { + return fmt.Errorf("live node identity is incomplete") + } + if want.Name != live.Name || want.Hostname != live.Hostname || want.UID != live.UID || want.ProviderID == "" || want.ProviderID != live.ProviderID { + return fmt.Errorf("node %q identity does not match immutable allowlist", live.Name) + } + return nil +} diff --git a/kafka-local-nvme/internal/repair/mount_linux.go b/kafka-local-nvme/internal/repair/mount_linux.go new file mode 100644 index 0000000..71913b5 --- /dev/null +++ b/kafka-local-nvme/internal/repair/mount_linux.go @@ -0,0 +1,87 @@ +//go:build linux + +package repair + +import ( + "fmt" + "os" + "sort" + "strings" +) + +func decodeMount(s string) string { + return strings.NewReplacer(`\040`, " ", `\011`, "\t", `\012`, "\n", `\134`, `\`).Replace(s) +} +func normalizedOptions(s string) []string { + m := map[string]bool{} + for _, v := range strings.Split(s, ",") { + if v != "" && v != "ro" && v != "rw" { + m[v] = true + } + } + out := make([]string, 0, len(m)) + for v := range m { + out = append(out, v) + } + sort.Strings(out) + return out +} +func sameOptions(a, b []string) bool { + a = append([]string(nil), a...) + b = append([]string(nil), b...) + sort.Strings(a) + sort.Strings(b) + return strings.Join(a, ",") == strings.Join(b, ",") +} + +// VerifyHostMount reads PID 1 mountinfo, so callers must use hostPID and a host /proc mount. +func VerifyHostMount(mountInfoPath string, want MountSignature) error { + for _, option := range want.Options { + if option == "ro" || option == "rw" { + return fmt.Errorf("configured mount options must omit access mode") + } + } + b, err := os.ReadFile(mountInfoPath) + if err != nil { + return err + } + var found []MountSignature + nested := false + for _, line := range strings.Split(strings.TrimSpace(string(b)), "\n") { + parts := strings.Split(line, " - ") + if len(parts) != 2 { + continue + } + left := strings.Fields(parts[0]) + right := strings.Fields(parts[1]) + if len(left) < 6 || len(right) < 3 { + continue + } + mountPoint := decodeMount(left[4]) + if strings.HasPrefix(mountPoint, HostRoot+"/") { + nested = true + continue + } + if mountPoint != HostRoot { + continue + } + rawOptions := left[5] + "," + right[2] + for _, option := range strings.Split(rawOptions, ",") { + if option == "ro" { + return fmt.Errorf("host mount is read-only") + } + } + found = append(found, MountSignature{Source: decodeMount(right[1]), FSType: right[0], Options: normalizedOptions(rawOptions)}) + } + if nested { + return fmt.Errorf("nested mount below %s", HostRoot) + } + if len(found) != 1 { + return fmt.Errorf("expected one host mount for %s, found %d", HostRoot, len(found)) + } + got := found[0] + if got.Source != want.Source || got.FSType != want.FSType || !sameOptions(got.Options, normalizedOptions(strings.Join(want.Options, ","))) { + return fmt.Errorf("host mount signature mismatch") + } + return nil +} diff --git a/kafka-local-nvme/internal/repair/repair_test.go b/kafka-local-nvme/internal/repair/repair_test.go new file mode 100644 index 0000000..f18aead --- /dev/null +++ b/kafka-local-nvme/internal/repair/repair_test.go @@ -0,0 +1,130 @@ +//go:build linux + +package repair + +import ( + "context" + "os" + "path/filepath" + "testing" +) + +type fakeReader struct { + pvs []PV + pv PV + pvc PVC + node NodeIdentity + change bool +} + +func (f *fakeReader) ListPVs(context.Context) ([]PV, error) { return f.pvs, nil } +func (f *fakeReader) GetPV(context.Context, string) (PV, error) { + if f.change { + x := f.pv + x.ClaimUID = "changed" + return x, nil + } + return f.pv, nil +} +func (f *fakeReader) GetPVC(context.Context, string, string) (PVC, error) { return f.pvc, nil } +func (f *fakeReader) GetNode(context.Context, string) (NodeIdentity, error) { return f.node, nil } + +func valid() (PV, PVC, NodeIdentity, Allowlist) { + n := NodeIdentity{"n", "host", "nodeuid", "provider"} + p := PV{Name: "pv", Phase: "Bound", VolumeMode: "local", StorageClass: StorageClass, LocalPath: "/mnt/local/kafka/kafka/data", ClaimNamespace: "kafka", ClaimName: "data", ClaimUID: "pvcuid", NodeHostname: "host", NodeOperator: "In", NodeAffinityValid: true} + c := PVC{"kafka", "data", "pvcuid", "pv", "Bound"} + return p, c, n, Allowlist{"n": n} +} +func TestCandidatesExactHierarchyAndIdentity(t *testing.T) { + p, c, n, a := valid() + f := &fakeReader{pvs: []PV{p}, pv: p, pvc: c, node: n} + got, err := Candidates(context.Background(), f, a, "n") + if err != nil || len(got) != 1 { + t.Fatalf("got %d, %v", len(got), err) + } + p.LocalPath = "/mnt/local/kafka/kafka/other" + f.pvs = []PV{p} + got, _ = Candidates(context.Background(), f, a, "n") + if len(got) != 0 { + t.Fatal("accepted wrong path") + } + p, c, n, a = valid() + f = &fakeReader{pvs: []PV{p}, pv: p, pvc: c, node: n} + got, _ = Candidates(context.Background(), f, a, "other") + if len(got) != 0 { + t.Fatal("accepted nonlocal node") + } +} +func TestVerifyIdentity(t *testing.T) { + _, _, n, a := valid() + if VerifyIdentity(a, n) != nil { + t.Fatal("valid identity rejected") + } + n.ProviderID = "other" + if VerifyIdentity(a, n) == nil { + t.Fatal("changed provider accepted") + } +} +func TestMountOptionsAndParsing(t *testing.T) { + d := t.TempDir() + f := filepath.Join(d, "mountinfo") + line := "1 0 8:1 / /mnt/local rw,nosuid - ext4 /dev/nvme0n1 rw,noatime\n" + if err := os.WriteFile(f, []byte(line), 0600); err != nil { + t.Fatal(err) + } + if err := VerifyHostMount(f, MountSignature{"/dev/nvme0n1", "ext4", []string{"noatime", "nosuid"}}); err != nil { + t.Fatal(err) + } + if VerifyHostMount(f, MountSignature{"/dev/nvme0n1", "ext4", []string{"ro"}}) == nil { + t.Fatal("bad options accepted") + } + if err := os.WriteFile(f, []byte("1 0 8:1 / /mnt/local ro - ext4 /dev/nvme0n1 rw\n"), 0600); err != nil { + t.Fatal(err) + } + if VerifyHostMount(f, MountSignature{"/dev/nvme0n1", "ext4", []string{}}) == nil { + t.Fatal("raw read-only mount accepted") + } +} + +func TestCandidateRejectsAmbiguousAffinity(t *testing.T) { + p, c, n, a := valid() + p.NodeAffinityValid = false + f := &fakeReader{pvs: []PV{p}, pv: p, pvc: c, node: n} + got, err := Candidates(context.Background(), f, a, "n") + if err != nil || len(got) != 0 { + t.Fatal("ambiguous affinity accepted") + } +} +func TestEnsureHierarchyRejectsSymlink(t *testing.T) { + root := t.TempDir() + if err := os.Symlink("/tmp", filepath.Join(root, "kafka")); err != nil { + t.Fatal(err) + } + if err := ensureHierarchy(root, filepath.Join(root, "kafka", "ns", "claim")); err == nil { + t.Fatal("symlink hierarchy accepted") + } +} +func TestEngineRevalidates(t *testing.T) { + p, c, n, a := valid() + f := &fakeReader{pvs: []PV{p}, pv: p, pvc: c, node: n, change: true} + d := t.TempDir() + m := filepath.Join(d, "mountinfo") + if err := os.WriteFile(m, []byte("1 0 8:1 / /mnt/local rw - ext4 /dev/nvme rw\n"), 0600); err != nil { + t.Fatal(err) + } + e := Engine{Reader: f, Allowlist: a, MountInfo: m, Mount: MountSignature{"/dev/nvme", "ext4", []string{"rw"}}} + if e.Repair(context.Background()) == nil { + t.Fatal("changed resource accepted") + } +} + +func TestLoadConfig(t *testing.T) { + path := filepath.Join(t.TempDir(), "config.json") + if err := os.WriteFile(path, []byte(`{"mountFilesystem":"ext4","mountSource":"/dev/nvme","mountOptions":["nodev"],"nodes":[{"name":"n","hostname":"host","uid":"nodeuid","providerID":"provider"}]}`), 0600); err != nil { + t.Fatal(err) + } + config, err := LoadConfig(path) + if err != nil || config.Nodes[0].Name != "n" { + t.Fatalf("config: %#v, %v", config, err) + } +} diff --git a/kafka-local-nvme/internal/repair/select.go b/kafka-local-nvme/internal/repair/select.go new file mode 100644 index 0000000..3a3f0b1 --- /dev/null +++ b/kafka-local-nvme/internal/repair/select.go @@ -0,0 +1,50 @@ +package repair + +import ( + "context" + "fmt" + "path" +) + +func ExpectedPath(namespace, name string) (string, error) { + if namespace == "" || name == "" || path.Base(namespace) != namespace || path.Base(name) != name { + return "", fmt.Errorf("invalid PVC identity") + } + return HostRoot + "/kafka/" + namespace + "/" + name, nil +} + +func Candidates(ctx context.Context, r Reader, allow Allowlist, localNode string) ([]Candidate, error) { + if localNode == "" { + return nil, fmt.Errorf("local node name is required") + } + pvs, err := r.ListPVs(ctx) + if err != nil { + return nil, err + } + var result []Candidate + for _, pv := range pvs { + if pv.Phase != "Bound" || pv.VolumeMode != "local" || pv.StorageClass != StorageClass || !pv.NodeAffinityValid || pv.NodeOperator != "In" || pv.ClaimNamespace == "" || pv.ClaimName == "" || pv.ClaimUID == "" || pv.NodeHostname == "" { + continue + } + expected, err := ExpectedPath(pv.ClaimNamespace, pv.ClaimName) + if err != nil || pv.LocalPath != expected { + continue + } + pvc, err := r.GetPVC(ctx, pv.ClaimNamespace, pv.ClaimName) + if err != nil || pvc.Phase != "Bound" || pvc.UID != pv.ClaimUID || pvc.VolumeName != pv.Name { + continue + } + var node NodeIdentity + for name, identity := range allow { + if identity.Hostname == pv.NodeHostname { + node, err = r.GetNode(ctx, name) + break + } + } + if err != nil || node.Name != localNode || VerifyIdentity(allow, node) != nil || node.Hostname != pv.NodeHostname { + continue + } + result = append(result, Candidate{PV: pv, PVC: pvc, Node: node}) + } + return result, nil +} diff --git a/kafka-local-nvme/internal/repair/types.go b/kafka-local-nvme/internal/repair/types.go new file mode 100644 index 0000000..d117a3b --- /dev/null +++ b/kafka-local-nvme/internal/repair/types.go @@ -0,0 +1,40 @@ +package repair + +import "context" + +const ( + StorageClass = "kafka-local" + HostRoot = "/mnt/local" +) + +type PV struct { + Name, Phase, VolumeMode, StorageClass, LocalPath, ClaimNamespace, ClaimName, ClaimUID, NodeHostname, NodeOperator string + NodeAffinityValid bool +} + +type PVC struct{ Namespace, Name, UID, VolumeName, Phase string } +type NodeIdentity struct{ Name, Hostname, UID, ProviderID string } + +// Reader is deliberately read-only. Implementations must never mutate Kubernetes state. +type Reader interface { + ListPVs(context.Context) ([]PV, error) + GetPV(context.Context, string) (PV, error) + GetPVC(context.Context, string, string) (PVC, error) + GetNode(context.Context, string) (NodeIdentity, error) +} + +type Allowlist map[string]NodeIdentity // keyed by node name +type Config struct { + MountFilesystem, MountSource string + MountOptions []string + Nodes []NodeIdentity +} +type Candidate struct { + PV PV + PVC PVC + Node NodeIdentity +} +type MountSignature struct { + Source, FSType string + Options []string +} diff --git a/kafka-local-nvme/profiles/five-node/brokers.yaml b/kafka-local-nvme/profiles/five-node/brokers.yaml new file mode 100644 index 0000000..d22a6f1 --- /dev/null +++ b/kafka-local-nvme/profiles/five-node/brokers.yaml @@ -0,0 +1,53 @@ +apiVersion: kafka.strimzi.io/v1 +kind: KafkaNodePool +metadata: + name: brokers + labels: + strimzi.io/cluster: kafka-local + app.kubernetes.io/part-of: kafka-local +spec: + replicas: 5 + roles: + - broker + # Strimzi creates one PVC per broker. The kafka-local StorageClass + # dynamically provisions a stable PVC-named local PV directory beneath + # /mnt/local/kafka on the node selected for the broker. + storage: + type: persistent-claim + size: 400Gi + class: kafka-local + deleteClaim: false + resources: + requests: + cpu: "4" + memory: 8Gi + limits: + cpu: "8" + memory: 8Gi + jvmOptions: + -Xms: 4g + -Xmx: 4g + template: + pod: + terminationGracePeriodSeconds: 300 + tolerations: + - key: dedicated + operator: Equal + value: kafka + effect: NoSchedule + affinity: + nodeAffinity: + requiredDuringSchedulingIgnoredDuringExecution: + nodeSelectorTerms: + - matchExpressions: + - key: kafka.local/kafka-local + operator: In + values: + - "true" + podAntiAffinity: + requiredDuringSchedulingIgnoredDuringExecution: + - topologyKey: kubernetes.io/hostname + labelSelector: + matchLabels: + strimzi.io/cluster: kafka-local + strimzi.io/broker-role: "true" diff --git a/kafka-local-nvme/profiles/five-node/controllers.yaml b/kafka-local-nvme/profiles/five-node/controllers.yaml new file mode 100644 index 0000000..b3c775a --- /dev/null +++ b/kafka-local-nvme/profiles/five-node/controllers.yaml @@ -0,0 +1,52 @@ +apiVersion: kafka.strimzi.io/v1 +kind: KafkaNodePool +metadata: + name: controllers + labels: + strimzi.io/cluster: kafka-local + app.kubernetes.io/part-of: kafka-local +spec: + replicas: 3 + roles: + - controller + # Controller metadata also uses local PVs. Losing multiple controller + # directories can lose KRaft quorum even when broker replicas are healthy. + storage: + type: persistent-claim + size: 20Gi + class: kafka-local + deleteClaim: false + resources: + requests: + cpu: "1" + memory: 2Gi + limits: + cpu: "2" + memory: 2Gi + jvmOptions: + -Xms: 1g + -Xmx: 1g + template: + pod: + terminationGracePeriodSeconds: 300 + tolerations: + - key: dedicated + operator: Equal + value: kafka + effect: NoSchedule + affinity: + nodeAffinity: + requiredDuringSchedulingIgnoredDuringExecution: + nodeSelectorTerms: + - matchExpressions: + - key: kafka.local/kafka-local + operator: In + values: + - "true" + podAntiAffinity: + requiredDuringSchedulingIgnoredDuringExecution: + - topologyKey: kubernetes.io/hostname + labelSelector: + matchLabels: + strimzi.io/cluster: kafka-local + strimzi.io/controller-role: "true" diff --git a/kafka-local-nvme/profiles/five-node/kafka.yaml b/kafka-local-nvme/profiles/five-node/kafka.yaml new file mode 100644 index 0000000..83b6246 --- /dev/null +++ b/kafka-local-nvme/profiles/five-node/kafka.yaml @@ -0,0 +1,45 @@ +apiVersion: kafka.strimzi.io/v1 +kind: Kafka +metadata: + name: kafka-local + labels: + app.kubernetes.io/name: kafka + app.kubernetes.io/instance: kafka-local + app.kubernetes.io/part-of: kafka-local + kafka.local/profile: five-node +spec: + kafka: + version: 4.3.0 + metadataVersion: "4.3" + listeners: + - name: plain + port: 9092 + type: internal + tls: false + - name: tls + port: 9093 + type: internal + tls: true + rack: + type: topology-label + topologyKey: kubernetes.io/hostname + # Strimzi generates one Kafka PodDisruptionBudget across all broker and + # controller node pools. The default is 1; it is explicit here so planned + # voluntary disruptions can never make more than one Kafka Pod unavailable. + template: + podDisruptionBudget: + maxUnavailable: 1 + config: + auto.create.topics.enable: false + min.insync.replicas: 3 + offsets.topic.replication.factor: 5 + transaction.state.log.replication.factor: 5 + transaction.state.log.min.isr: 3 + unclean.leader.election.enable: false + entityOperator: + topicOperator: {} + userOperator: {} + cruiseControl: {} + kafkaExporter: + groupRegex: ".*" + topicRegex: ".*" diff --git a/kafka-local-nvme/profiles/five-node/kustomization.yaml b/kafka-local-nvme/profiles/five-node/kustomization.yaml new file mode 100644 index 0000000..bb03fae --- /dev/null +++ b/kafka-local-nvme/profiles/five-node/kustomization.yaml @@ -0,0 +1,8 @@ +apiVersion: kustomize.config.k8s.io/v1beta1 +kind: Kustomization +namespace: kafka +resources: + - kafka.yaml + - controllers.yaml + - brokers.yaml + - topic.yaml diff --git a/kafka-local-nvme/profiles/five-node/topic.yaml b/kafka-local-nvme/profiles/five-node/topic.yaml new file mode 100644 index 0000000..6da281f --- /dev/null +++ b/kafka-local-nvme/profiles/five-node/topic.yaml @@ -0,0 +1,15 @@ +apiVersion: kafka.strimzi.io/v1 +kind: KafkaTopic +metadata: + name: kafka-local-topic + labels: + strimzi.io/cluster: kafka-local + app.kubernetes.io/part-of: kafka-local + kafka.local/profile: five-node +spec: + partitions: 10 + replicas: 5 + config: + cleanup.policy: delete + min.insync.replicas: 3 + retention.ms: 86400000 diff --git a/kafka-local-nvme/profiles/two-node/brokers.yaml b/kafka-local-nvme/profiles/two-node/brokers.yaml new file mode 100644 index 0000000..449955a --- /dev/null +++ b/kafka-local-nvme/profiles/two-node/brokers.yaml @@ -0,0 +1,53 @@ +apiVersion: kafka.strimzi.io/v1 +kind: KafkaNodePool +metadata: + name: brokers + labels: + strimzi.io/cluster: kafka-local + app.kubernetes.io/part-of: kafka-local +spec: + replicas: 2 + roles: + - broker + # Strimzi creates one PVC per broker. The kafka-local StorageClass + # dynamically provisions a stable PVC-named local PV directory beneath + # /mnt/local/kafka on the node selected for the broker. + storage: + type: persistent-claim + size: 400Gi + class: kafka-local + deleteClaim: false + resources: + requests: + cpu: "2" + memory: 4Gi + limits: + cpu: "4" + memory: 4Gi + jvmOptions: + -Xms: 2g + -Xmx: 2g + template: + pod: + terminationGracePeriodSeconds: 300 + tolerations: + - key: dedicated + operator: Equal + value: kafka + effect: NoSchedule + affinity: + nodeAffinity: + requiredDuringSchedulingIgnoredDuringExecution: + nodeSelectorTerms: + - matchExpressions: + - key: kafka.local/kafka-local + operator: In + values: + - "true" + podAntiAffinity: + requiredDuringSchedulingIgnoredDuringExecution: + - topologyKey: kubernetes.io/hostname + labelSelector: + matchLabels: + strimzi.io/cluster: kafka-local + strimzi.io/broker-role: "true" diff --git a/kafka-local-nvme/profiles/two-node/controllers.yaml b/kafka-local-nvme/profiles/two-node/controllers.yaml new file mode 100644 index 0000000..9bf4d20 --- /dev/null +++ b/kafka-local-nvme/profiles/two-node/controllers.yaml @@ -0,0 +1,54 @@ +apiVersion: kafka.strimzi.io/v1 +kind: KafkaNodePool +metadata: + name: controllers + labels: + strimzi.io/cluster: kafka-local + app.kubernetes.io/part-of: kafka-local +spec: + replicas: 3 + roles: + - controller + # Controller metadata also uses local PVs. Losing multiple controller + # directories can lose KRaft quorum even when broker replicas are healthy. + storage: + type: persistent-claim + size: 20Gi + class: kafka-local + deleteClaim: false + resources: + requests: + cpu: "500m" + memory: 1Gi + limits: + cpu: "1" + memory: 2Gi + jvmOptions: + -Xms: 1g + -Xmx: 1g + template: + pod: + terminationGracePeriodSeconds: 300 + tolerations: + - key: dedicated + operator: Equal + value: kafka + effect: NoSchedule + affinity: + nodeAffinity: + requiredDuringSchedulingIgnoredDuringExecution: + nodeSelectorTerms: + - matchExpressions: + - key: kafka.local/kafka-local + operator: In + values: + - "true" + podAntiAffinity: + preferredDuringSchedulingIgnoredDuringExecution: + - weight: 100 + podAffinityTerm: + topologyKey: kubernetes.io/hostname + labelSelector: + matchLabels: + strimzi.io/cluster: kafka-local + strimzi.io/controller-role: "true" diff --git a/kafka-local-nvme/profiles/two-node/kafka.yaml b/kafka-local-nvme/profiles/two-node/kafka.yaml new file mode 100644 index 0000000..d277f0f --- /dev/null +++ b/kafka-local-nvme/profiles/two-node/kafka.yaml @@ -0,0 +1,44 @@ +apiVersion: kafka.strimzi.io/v1 +kind: Kafka +metadata: + name: kafka-local + labels: + app.kubernetes.io/name: kafka + app.kubernetes.io/instance: kafka-local + app.kubernetes.io/part-of: kafka-local + kafka.local/profile: two-node +spec: + kafka: + version: 4.3.0 + metadataVersion: "4.3" + listeners: + - name: plain + port: 9092 + type: internal + tls: false + - name: tls + port: 9093 + type: internal + tls: true + rack: + type: topology-label + topologyKey: kubernetes.io/hostname + # Strimzi generates one Kafka PodDisruptionBudget across all broker and + # controller node pools. The default is 1; it is explicit here so planned + # voluntary disruptions can never make more than one Kafka Pod unavailable. + template: + podDisruptionBudget: + maxUnavailable: 1 + config: + auto.create.topics.enable: false + min.insync.replicas: 1 + offsets.topic.replication.factor: 2 + transaction.state.log.replication.factor: 2 + transaction.state.log.min.isr: 1 + unclean.leader.election.enable: false + entityOperator: + topicOperator: {} + userOperator: {} + kafkaExporter: + groupRegex: ".*" + topicRegex: ".*" diff --git a/kafka-local-nvme/profiles/two-node/kustomization.yaml b/kafka-local-nvme/profiles/two-node/kustomization.yaml new file mode 100644 index 0000000..bb03fae --- /dev/null +++ b/kafka-local-nvme/profiles/two-node/kustomization.yaml @@ -0,0 +1,8 @@ +apiVersion: kustomize.config.k8s.io/v1beta1 +kind: Kustomization +namespace: kafka +resources: + - kafka.yaml + - controllers.yaml + - brokers.yaml + - topic.yaml diff --git a/kafka-local-nvme/profiles/two-node/topic.yaml b/kafka-local-nvme/profiles/two-node/topic.yaml new file mode 100644 index 0000000..837e266 --- /dev/null +++ b/kafka-local-nvme/profiles/two-node/topic.yaml @@ -0,0 +1,15 @@ +apiVersion: kafka.strimzi.io/v1 +kind: KafkaTopic +metadata: + name: kafka-local-topic + labels: + strimzi.io/cluster: kafka-local + app.kubernetes.io/part-of: kafka-local + kafka.local/profile: two-node +spec: + partitions: 10 + replicas: 2 + config: + cleanup.policy: delete + min.insync.replicas: 1 + retention.ms: 86400000 diff --git a/kafka-local-nvme/provisioner/local-path/config-map.yaml b/kafka-local-nvme/provisioner/local-path/config-map.yaml new file mode 100644 index 0000000..6aca924 --- /dev/null +++ b/kafka-local-nvme/provisioner/local-path/config-map.yaml @@ -0,0 +1,60 @@ +apiVersion: v1 +kind: ConfigMap +metadata: + name: local-path-config + namespace: local-path-storage + labels: + app.kubernetes.io/name: local-path-provisioner + kafka.local/component: storage-provisioner +data: + config.json: |- + { + "nodePathMap": [ + { + "node": "DEFAULT_PATH_FOR_NON_LISTED_NODES", + "paths": [ + "/mnt/local/kafka" + ] + } + ] + } + # The helper Pod runs this automatically when a PVC is provisioned. + setup: |- + #!/bin/sh + set -eu + mkdir -m 0770 -p "$VOL_DIR" + chown 1001:0 "$VOL_DIR" + chmod 0770 "$VOL_DIR" + # reclaimPolicy is Retain, so normal PVC deletion does not invoke teardown. + # Keep a path guard in case the policy is deliberately changed in future. + teardown: |- + #!/bin/sh + set -eu + case "$VOL_DIR" in + /mnt/local/kafka/*) + rm -rf -- "$VOL_DIR" + ;; + *) + echo "Refusing to remove unexpected path: $VOL_DIR" >&2 + exit 1 + ;; + esac + helperPod.yaml: |- + apiVersion: v1 + kind: Pod + metadata: + name: helper-pod + spec: + priorityClassName: system-node-critical + tolerations: + - key: node.kubernetes.io/disk-pressure + operator: Exists + effect: NoSchedule + - key: dedicated + operator: Equal + value: kafka + effect: NoSchedule + containers: + - name: helper-pod + image: docker.io/library/busybox:1.36.1 + imagePullPolicy: IfNotPresent diff --git a/kafka-local-nvme/provisioner/local-path/deployment.yaml b/kafka-local-nvme/provisioner/local-path/deployment.yaml new file mode 100644 index 0000000..2dc192b --- /dev/null +++ b/kafka-local-nvme/provisioner/local-path/deployment.yaml @@ -0,0 +1,62 @@ +# Pinned to Rancher Local Path Provisioner v0.0.36. +apiVersion: apps/v1 +kind: Deployment +metadata: + name: local-path-provisioner + namespace: local-path-storage + labels: + app.kubernetes.io/name: local-path-provisioner + app.kubernetes.io/version: 0.0.36 + kafka.local/component: storage-provisioner +spec: + replicas: 1 + selector: + matchLabels: + app.kubernetes.io/name: local-path-provisioner + template: + metadata: + labels: + app.kubernetes.io/name: local-path-provisioner + app.kubernetes.io/version: 0.0.36 + kafka.local/component: storage-provisioner + spec: + serviceAccountName: local-path-provisioner-service-account + containers: + - name: local-path-provisioner + image: docker.io/rancher/local-path-provisioner:v0.0.36 + imagePullPolicy: IfNotPresent + command: + - local-path-provisioner + - --debug + - start + - --config + - /etc/config/config.json + env: + - name: POD_NAMESPACE + valueFrom: + fieldRef: + fieldPath: metadata.namespace + - name: CONFIG_MOUNT_PATH + value: /etc/config/ + resources: + requests: + cpu: 10m + memory: 32Mi + limits: + cpu: 200m + memory: 128Mi + securityContext: + allowPrivilegeEscalation: false + capabilities: + drop: + - ALL + runAsNonRoot: true + runAsUser: 65534 + volumeMounts: + - name: config-volume + mountPath: /etc/config/ + readOnly: true + volumes: + - name: config-volume + configMap: + name: local-path-config diff --git a/kafka-local-nvme/provisioner/local-path/kustomization.yaml b/kafka-local-nvme/provisioner/local-path/kustomization.yaml new file mode 100644 index 0000000..3bc830f --- /dev/null +++ b/kafka-local-nvme/provisioner/local-path/kustomization.yaml @@ -0,0 +1,8 @@ +apiVersion: kustomize.config.k8s.io/v1beta1 +kind: Kustomization +resources: + - namespace.yaml + - rbac.yaml + - config-map.yaml + - deployment.yaml + - storage-class.yaml diff --git a/kafka-local-nvme/provisioner/local-path/namespace.yaml b/kafka-local-nvme/provisioner/local-path/namespace.yaml new file mode 100644 index 0000000..6b18b12 --- /dev/null +++ b/kafka-local-nvme/provisioner/local-path/namespace.yaml @@ -0,0 +1,7 @@ +apiVersion: v1 +kind: Namespace +metadata: + name: local-path-storage + labels: + app.kubernetes.io/name: local-path-provisioner + kafka.local/component: storage-provisioner diff --git a/kafka-local-nvme/provisioner/local-path/rbac.yaml b/kafka-local-nvme/provisioner/local-path/rbac.yaml new file mode 100644 index 0000000..fa4a296 --- /dev/null +++ b/kafka-local-nvme/provisioner/local-path/rbac.yaml @@ -0,0 +1,113 @@ +# Derived from Rancher Local Path Provisioner v0.0.36: +# https://github.com/rancher/local-path-provisioner +apiVersion: v1 +kind: ServiceAccount +metadata: + name: local-path-provisioner-service-account + namespace: local-path-storage + labels: + app.kubernetes.io/name: local-path-provisioner + kafka.local/component: storage-provisioner +--- +apiVersion: rbac.authorization.k8s.io/v1 +kind: Role +metadata: + name: local-path-provisioner-role + namespace: local-path-storage + labels: + app.kubernetes.io/name: local-path-provisioner + kafka.local/component: storage-provisioner +rules: + - apiGroups: + - "" + resources: + - pods + verbs: + - get + - list + - watch + - create + - patch + - update + - delete +--- +apiVersion: rbac.authorization.k8s.io/v1 +kind: ClusterRole +metadata: + name: local-path-provisioner-role + labels: + app.kubernetes.io/name: local-path-provisioner + kafka.local/component: storage-provisioner +rules: + - apiGroups: + - "" + resources: + - nodes + - persistentvolumeclaims + - configmaps + - pods + - pods/log + verbs: + - get + - list + - watch + - apiGroups: + - "" + resources: + - persistentvolumes + verbs: + - get + - list + - watch + - create + - patch + - update + - delete + - apiGroups: + - "" + resources: + - events + verbs: + - create + - patch + - apiGroups: + - storage.k8s.io + resources: + - storageclasses + verbs: + - get + - list + - watch +--- +apiVersion: rbac.authorization.k8s.io/v1 +kind: RoleBinding +metadata: + name: local-path-provisioner-bind + namespace: local-path-storage + labels: + app.kubernetes.io/name: local-path-provisioner + kafka.local/component: storage-provisioner +roleRef: + apiGroup: rbac.authorization.k8s.io + kind: Role + name: local-path-provisioner-role +subjects: + - kind: ServiceAccount + name: local-path-provisioner-service-account + namespace: local-path-storage +--- +apiVersion: rbac.authorization.k8s.io/v1 +kind: ClusterRoleBinding +metadata: + name: local-path-provisioner-bind + labels: + app.kubernetes.io/name: local-path-provisioner + kafka.local/component: storage-provisioner +roleRef: + apiGroup: rbac.authorization.k8s.io + kind: ClusterRole + name: local-path-provisioner-role +subjects: + - kind: ServiceAccount + name: local-path-provisioner-service-account + namespace: local-path-storage diff --git a/kafka-local-nvme/provisioner/local-path/repair/daemonset.template.yaml b/kafka-local-nvme/provisioner/local-path/repair/daemonset.template.yaml new file mode 100644 index 0000000..c9f1bae --- /dev/null +++ b/kafka-local-nvme/provisioner/local-path/repair/daemonset.template.yaml @@ -0,0 +1,73 @@ +apiVersion: v1 +kind: ConfigMap +metadata: + name: __CONFIG_MAP_NAME__ + namespace: kafka +immutable: true +data: + config.json: |- +__CONFIG_JSON_INDENTED__ +--- +apiVersion: rbac.authorization.k8s.io/v1 +kind: ClusterRole +metadata: + name: kafka-local-pv-repair-reader +rules: + - apiGroups: [""] + resources: ["persistentvolumes"] + verbs: ["get", "list"] + - apiGroups: [""] + resources: ["nodes"] + resourceNames: __NODE_NAMES_JSON__ + verbs: ["get"] +--- +apiVersion: rbac.authorization.k8s.io/v1 +kind: ClusterRoleBinding +metadata: + name: kafka-local-pv-repair-reader +subjects: + - kind: ServiceAccount + name: kafka-local-pv-repair + namespace: kafka +roleRef: + apiGroup: rbac.authorization.k8s.io + kind: ClusterRole + name: kafka-local-pv-repair-reader +--- +apiVersion: apps/v1 +kind: DaemonSet +metadata: + name: kafka-local-pv-repair + namespace: kafka +spec: + selector: {matchLabels: {app.kubernetes.io/name: kafka-local-pv-repair}} + template: + metadata: + labels: {app.kubernetes.io/name: kafka-local-pv-repair} + annotations: {kafka.local/repair-config: __CONFIG_HASH__} + spec: + hostPID: true + serviceAccountName: kafka-local-pv-repair + nodeSelector: {kafka.local/kafka-local: "true"} + containers: + - name: repair + image: __IMAGE__ + env: + - {name: REPAIR_CONFIG_PATH, value: /etc/repair/config.json} + - name: NODE_NAME + valueFrom: {fieldRef: {fieldPath: spec.nodeName}} + resources: {requests: {cpu: 25m, memory: 64Mi}, limits: {cpu: 100m, memory: 128Mi}} + securityContext: + readOnlyRootFilesystem: true + allowPrivilegeEscalation: false + runAsUser: 0 + capabilities: {drop: ["ALL"], add: ["CHOWN", "DAC_OVERRIDE"]} + seccompProfile: {type: RuntimeDefault} + volumeMounts: + - {name: config, mountPath: /etc/repair, readOnly: true} + - {name: host-mnt-local, mountPath: /mnt/local} + volumes: + - name: config + configMap: {name: __CONFIG_MAP_NAME__} + - name: host-mnt-local + hostPath: {path: /mnt/local, type: Directory} diff --git a/kafka-local-nvme/provisioner/local-path/repair/rbac.yaml b/kafka-local-nvme/provisioner/local-path/repair/rbac.yaml new file mode 100644 index 0000000..df9d375 --- /dev/null +++ b/kafka-local-nvme/provisioner/local-path/repair/rbac.yaml @@ -0,0 +1,23 @@ +apiVersion: rbac.authorization.k8s.io/v1 +kind: Role +metadata: + name: kafka-local-pv-repair + namespace: kafka +rules: + - apiGroups: [""] + resources: ["persistentvolumeclaims"] + verbs: ["get"] +--- +apiVersion: rbac.authorization.k8s.io/v1 +kind: RoleBinding +metadata: + name: kafka-local-pv-repair + namespace: kafka +subjects: + - kind: ServiceAccount + name: kafka-local-pv-repair + namespace: kafka +roleRef: + apiGroup: rbac.authorization.k8s.io + kind: Role + name: kafka-local-pv-repair diff --git a/kafka-local-nvme/provisioner/local-path/repair/service-account.yaml b/kafka-local-nvme/provisioner/local-path/repair/service-account.yaml new file mode 100644 index 0000000..1b90473 --- /dev/null +++ b/kafka-local-nvme/provisioner/local-path/repair/service-account.yaml @@ -0,0 +1,5 @@ +apiVersion: v1 +kind: ServiceAccount +metadata: + name: kafka-local-pv-repair + namespace: kafka diff --git a/kafka-local-nvme/provisioner/local-path/storage-class.yaml b/kafka-local-nvme/provisioner/local-path/storage-class.yaml new file mode 100644 index 0000000..66ca166 --- /dev/null +++ b/kafka-local-nvme/provisioner/local-path/storage-class.yaml @@ -0,0 +1,18 @@ +apiVersion: storage.k8s.io/v1 +kind: StorageClass +metadata: + name: kafka-local + labels: + app.kubernetes.io/name: local-path-provisioner + kafka.local/component: storage-provisioner + annotations: + # Create Kubernetes local PVs with topology tied to the selected node. + defaultVolumeType: local +provisioner: rancher.io/local-path +parameters: + # Keep the directory stable for the Strimzi PVC name, matching the + # ordinal-style storage identity expected from a stateful workload. + nodePath: /mnt/local/kafka + pathPattern: "{{ .PVC.Namespace }}/{{ .PVC.Name }}/" +reclaimPolicy: Retain +volumeBindingMode: WaitForFirstConsumer diff --git a/kafka-local-nvme/scripts/render-kafka-local-pv-repair-manifest.sh b/kafka-local-nvme/scripts/render-kafka-local-pv-repair-manifest.sh new file mode 100755 index 0000000..f0d2bb6 --- /dev/null +++ b/kafka-local-nvme/scripts/render-kafka-local-pv-repair-manifest.sh @@ -0,0 +1,37 @@ +#!/usr/bin/env bash +set -euo pipefail +root="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)" +image= fs= source= options= nodes= +while [[ $# -gt 0 ]]; do case "$1" in --image) image="${2:-}";shift 2;;--mount-fs) fs="${2:-}";shift 2;;--mount-source) source="${2:-}";shift 2;;--mount-options) options="${2:-}";shift 2;;--nodes) nodes="${2:-}";shift 2;;*) printf 'error: unknown argument %s\n' "$1" >&2;exit 2;;esac;done +python3 - "$root/provisioner/local-path/repair/daemonset.template.yaml" "$image" "$fs" "$source" "$options" "$nodes" <<'PY' +import hashlib,json,re,sys +template,image,fs,source,options,nodes=sys.argv[1:] +def fail(msg): raise SystemExit('error: '+msg) +if not re.fullmatch(r'[a-z0-9][a-z0-9._/-]*@sha256:[a-f0-9]{64}',image): fail('image must be repository@sha256:64hex') +if not re.fullmatch(r'[A-Za-z0-9._+-]+',fs): fail('invalid mount filesystem') +if not source.startswith('/') or '\n' in source or '\r' in source: fail('mount source must be an absolute path') +parts=options.split(',') +if not parts or any(not re.fullmatch(r'[A-Za-z0-9._=-]+',x) or x in ('ro','rw') for x in parts) or len(set(parts))!=len(parts) or parts!=sorted(parts): fail('mount options must be canonical, safe, unique, and omit mode') +out=[]; names=set(); hosts=set() +try: lines=open(nodes,encoding='utf-8').read().splitlines() +except OSError as e: fail(str(e)) +for line in lines: + fields=line.split('\t') + if len(fields)!=4: fail('nodes must be namehostnameUIDproviderID') + name,host,uid,provider=fields + if not re.fullmatch(r'[a-z0-9](?:[a-z0-9.-]*[a-z0-9])?',name) or not re.fullmatch(r'[a-z0-9](?:[a-z0-9.-]*[a-z0-9])?',host) or not re.fullmatch(r'[0-9a-f-]{8,}',uid) or not provider or '\n' in provider or '\r' in provider: fail('invalid node identity') + if name in names or host in hosts: fail('duplicate node name or hostname') + names.add(name);hosts.add(host);out.append({'name':name,'hostname':host,'uid':uid,'providerID':provider}) +if not out: fail('nodes TSV is empty') +config={'mountFilesystem':fs,'mountOptions':parts,'mountSource':source,'nodes':out} +config_json=json.dumps(config,separators=(',',':'),sort_keys=True) +digest=hashlib.sha256(config_json.encode()).hexdigest(); name='kafka-local-pv-repair-config-'+digest[:16] +s=open(template).read(); values={'CONFIG_JSON_INDENTED':' '+config_json,'CONFIG_MAP_NAME':name,'NODE_NAMES_JSON':json.dumps([x['name'] for x in out],separators=(',',':')),'CONFIG_HASH':digest,'IMAGE':image} +for key,value in values.items(): s=s.replace('__'+key+'__',value) +repair_dir=__import__('os').path.dirname(template) +print(open(__import__('os').path.join(repair_dir,'service-account.yaml')).read().rstrip()) +print('---') +print(open(__import__('os').path.join(repair_dir,'rbac.yaml')).read().rstrip()) +print('---') +print(s,end='') +PY diff --git a/kafka-local-nvme/scripts/smoke-test.sh b/kafka-local-nvme/scripts/smoke-test.sh new file mode 100755 index 0000000..9d32e6a --- /dev/null +++ b/kafka-local-nvme/scripts/smoke-test.sh @@ -0,0 +1,93 @@ +#!/usr/bin/env bash +set -euo pipefail + +namespace="${NAMESPACE:-kafka}" +cluster="${KAFKA_CLUSTER:-kafka-local}" +topic="${KAFKA_TOPIC:-kafka-local-topic}" +message_count="${MESSAGE_COUNT:-10}" +kafka_image="${KAFKA_IMAGE:-quay.io/strimzi/kafka:1.1.0-kafka-4.3.0}" +bootstrap="${cluster}-kafka-bootstrap:9092" +run_id="$(date -u +%Y%m%dT%H%M%SZ)-$$" +producer_pod="kafka-local-producer-$$" +consumer_pod="kafka-local-consumer-$$" +consumer_output="$(mktemp)" +expected_messages="$(mktemp)" +consumed_messages="$(mktemp)" + +cleanup() { + kubectl --namespace "$namespace" delete pod \ + "$producer_pod" "$consumer_pod" \ + --ignore-not-found \ + --wait=false >/dev/null 2>&1 || true + rm -f "$consumer_output" "$expected_messages" "$consumed_messages" +} +trap cleanup EXIT + +if [[ ! "$message_count" =~ ^[1-9][0-9]*$ ]]; then + printf 'MESSAGE_COUNT must be a positive integer.\n' >&2 + exit 2 +fi + +kubectl --namespace "$namespace" wait \ + --for=condition=Ready \ + "kafkatopic/${topic}" \ + --timeout=300s + +printf 'Producing %s messages with run ID %s\n' "$message_count" "$run_id" + +for message_number in $(seq 1 "$message_count"); do + printf '%s\n' \ + "${run_id}|message=${message_number}|timestamp=$(date -u +%FT%TZ)" +done > "$expected_messages" + +kubectl run "$producer_pod" \ + --namespace "$namespace" \ + --image "$kafka_image" \ + --restart Never \ + --attach=true \ + --pod-running-timeout=5m \ + --overrides='{"spec":{"tolerations":[{"key":"dedicated","operator":"Equal","value":"kafka","effect":"NoSchedule"}]}}' \ + -i < "$expected_messages" \ + --command -- \ + bin/kafka-console-producer.sh \ + --bootstrap-server "$bootstrap" \ + --topic "$topic" \ + --command-property acks=all + +kubectl --namespace "$namespace" wait \ + --for=jsonpath='{.status.phase}'=Succeeded \ + "pod/${producer_pod}" \ + --timeout=60s + +printf 'Reading messages back from %s\n' "$topic" + +kubectl run "$consumer_pod" \ + --namespace "$namespace" \ + --image "$kafka_image" \ + --restart Never \ + --attach=true \ + --pod-running-timeout=5m \ + --overrides='{"spec":{"tolerations":[{"key":"dedicated","operator":"Equal","value":"kafka","effect":"NoSchedule"}]}}' \ + -i \ + --command -- \ + bin/kafka-console-consumer.sh \ + --bootstrap-server "$bootstrap" \ + --topic "$topic" \ + --from-beginning \ + --timeout-ms 15000 \ + >"$consumer_output" 2>/dev/null || true + +grep -F "$run_id" "$consumer_output" > "$consumed_messages" || true + +if ! diff --brief \ + <(LC_ALL=C sort "$expected_messages") \ + <(LC_ALL=C sort "$consumed_messages") >/dev/null; then + printf 'Exact readback failed for run ID %s.\n' "$run_id" >&2 + diff -u \ + <(LC_ALL=C sort "$expected_messages") \ + <(LC_ALL=C sort "$consumed_messages") >&2 || true + exit 1 +fi + +printf 'Successfully produced and exactly read %s messages for %s\n' \ + "$message_count" "$run_id" diff --git a/kafka-local-nvme/scripts/validate.sh b/kafka-local-nvme/scripts/validate.sh new file mode 100755 index 0000000..cd4e1ba --- /dev/null +++ b/kafka-local-nvme/scripts/validate.sh @@ -0,0 +1,101 @@ +#!/usr/bin/env bash +# Validate the checked-in deployment manifests without a cluster. +set -euo pipefail + +script_directory="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" +root_directory="$(cd "$script_directory/.." && pwd)" +temporary_directory="$(mktemp -d)" +trap 'rm -rf "$temporary_directory"' EXIT + +command -v kubectl >/dev/null 2>&1 || { + printf 'kubectl is required for Kustomize rendering.\n' >&2 + exit 1 +} + +fail() { + printf 'Validation error: %s\n' "$1" >&2 + exit 1 +} + +for script in "$script_directory"/*.sh; do + bash -n "$script" || fail "invalid Bash syntax: ${script#$root_directory/}" +done +bash -n "$root_directory/tests/render-kafka-local-pv-repair-manifest-test.sh" || fail 'invalid renderer test syntax' +"$root_directory/tests/render-kafka-local-pv-repair-manifest-test.sh" || fail 'renderer test failed' + +( + cd "$root_directory" + GOCACHE="$temporary_directory/go-build" go test ./... + GOCACHE="$temporary_directory/go-build" GOOS=linux GOARCH=amd64 CGO_ENABLED=0 go test -c ./internal/repair -o "$temporary_directory/repair-linux.test" + GOCACHE="$temporary_directory/go-build" GOOS=linux GOARCH=amd64 CGO_ENABLED=0 go vet ./internal/repair ./cmd/kafka-local-pv-repair +) || fail 'Go validation failed' + +provisioner="$temporary_directory/provisioner.yaml" +two_node="$temporary_directory/two-node.yaml" +five_node="$temporary_directory/five-node.yaml" +kubectl kustomize "$root_directory/provisioner/local-path" > "$provisioner" +kubectl kustomize "$root_directory/profiles/two-node" > "$two_node" +kubectl kustomize "$root_directory/profiles/five-node" > "$five_node" + +require() { + local file="$1" pattern="$2" description="$3" + grep -Eq -- "$pattern" "$file" || fail "$description" +} + +require_count() { + local file="$1" pattern="$2" expected="$3" description="$4" actual + actual="$(grep -Ec -- "$pattern" "$file" || true)" + [[ "$actual" -eq "$expected" ]] || fail "$description (found $actual, expected $expected)." +} + +require_resource() { + local file="$1" kind="$2" pattern="$3" description="$4" + awk -v kind="$kind" -v pattern="$pattern" ' + function check() { + if (resource_kind == kind && resource ~ pattern) found = 1 + } + /^---$/ { check(); resource_kind = ""; resource = ""; next } + { resource = resource $0 "\n" } + $0 == "kind: " kind { resource_kind = kind } + END { check(); exit(found ? 0 : 1) } + ' "$file" || fail "$description" +} + +require_node_pool() { + local file="$1" role="$2" replicas="$3" size="$4" description="$5" + awk -v role="$role" -v replicas="$replicas" -v size="$size" ' + function check() { + if (kind == "KafkaNodePool" && has_role && has_replicas && has_size && has_class && has_delete_claim) found = 1 + } + /^---$/ { check(); kind = ""; has_role = has_replicas = has_size = has_class = has_delete_claim = 0; next } + $0 == "kind: KafkaNodePool" { kind = "KafkaNodePool" } + kind == "KafkaNodePool" && $0 == " - " role { has_role = 1 } + kind == "KafkaNodePool" && $0 == " replicas: " replicas { has_replicas = 1 } + kind == "KafkaNodePool" && $0 == " size: " size { has_size = 1 } + kind == "KafkaNodePool" && $0 == " class: kafka-local" { has_class = 1 } + kind == "KafkaNodePool" && $0 == " deleteClaim: false" { has_delete_claim = 1 } + END { check(); exit(found ? 0 : 1) } + ' "$file" || fail "$description" +} + +require "$provisioner" 'nodePath: /mnt/local/kafka' 'StorageClass must use /mnt/local/kafka.' +require "$provisioner" 'reclaimPolicy: Retain' 'StorageClass must retain PVs.' +require "$provisioner" 'volumeBindingMode: WaitForFirstConsumer' 'StorageClass must wait for a consumer.' +require "$provisioner" 'defaultVolumeType: local' 'StorageClass must create local PVs.' +require "$provisioner" '"/mnt/local/kafka"' 'Provisioner ConfigMap must restrict helper Pods to /mnt/local/kafka.' + +validate_profile() { + local file="$1" brokers="$2" topic_replicas="$3" min_isr="$4" profile="$5" + require_count "$file" '^kind: KafkaNodePool$' 2 "$profile must define broker and controller pools" + require_count "$file" '^kind: KafkaTopic$' 1 "$profile must define one topic" + require_node_pool "$file" broker "$brokers" 400Gi "$profile broker pool is incorrect." + require_node_pool "$file" controller 3 20Gi "$profile controller pool is incorrect." + require_resource "$file" KafkaTopic "replicas: ${topic_replicas}" "$profile topic replication factor is incorrect." + require_resource "$file" KafkaTopic "min.insync.replicas: ${min_isr}" "$profile topic min ISR is incorrect." + require "$file" 'key: kafka.local/kafka-local' "$profile must select labelled Kafka nodes." + require "$file" 'topologyKey: kubernetes.io/hostname' "$profile must use hostname placement." +} + +validate_profile "$two_node" 2 2 1 two-node +validate_profile "$five_node" 5 5 3 five-node +printf 'Manifest validation passed.\n' diff --git a/kafka-local-nvme/tests/render-kafka-local-pv-repair-manifest-test.sh b/kafka-local-nvme/tests/render-kafka-local-pv-repair-manifest-test.sh new file mode 100755 index 0000000..308ece5 --- /dev/null +++ b/kafka-local-nvme/tests/render-kafka-local-pv-repair-manifest-test.sh @@ -0,0 +1,14 @@ +#!/usr/bin/env bash +set -euo pipefail +root="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)" +r="$root/scripts/render-kafka-local-pv-repair-manifest.sh" +t="$(mktemp -d)";trap 'rm -rf "$t"' EXIT +printf 'node-a\thost-a\t12345678-abcd\tcloud://node-a\n' > "$t/nodes" +a=(--image registry.example/repair@sha256:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa --mount-fs ext4 --mount-source /dev/nvme0n1 --mount-options nodev,nosuid --nodes "$t/nodes") +"$r" "${a[@]}" > "$t/out" +for kind in ServiceAccount Role RoleBinding ConfigMap ClusterRole ClusterRoleBinding DaemonSet; do grep -q "kind: $kind" "$t/out"; done +awk '/^kind: ClusterRoleBinding$/{in_binding=1} in_binding && /^roleRef:/{in_role_ref=1} in_role_ref && /^ name: kafka-local-pv-repair-reader$/{found=1} END{exit(found ? 0 : 1)}' "$t/out" +grep -q 'immutable: true' "$t/out";grep -q 'hostPID: true' "$t/out";grep -q 'readOnlyRootFilesystem: true' "$t/out";grep -q 'type: RuntimeDefault' "$t/out";grep -q 'resourceNames: \["node-a"\]' "$t/out";grep -q 'hostPath: {path: /mnt/local, type: Directory}' "$t/out";grep -q 'verbs: \["get", "list"\]' "$t/out";grep -q 'resources: \["persistentvolumeclaims"\]' "$root/provisioner/local-path/repair/rbac.yaml";! grep -Eq 'verbs:.*(create|update|patch|delete|watch)' "$t/out";! grep -Eq '(ports:|readinessProbe:|livenessProbe:|updateStrategy:)' "$t/out" +! "$r" --image bad --mount-fs ext4 --mount-source /dev/x --mount-options nosuid --nodes "$t/nodes" >/dev/null 2>&1 +! "$r" --image "${a[1]}" --mount-fs ext4 --mount-source /dev/x --mount-options nosuid,nosuid --nodes "$t/nodes" >/dev/null 2>&1 +printf 'node-a\thost-b\t12345678-abcd\tcloud://node-b\n' >> "$t/nodes";! "$r" "${a[@]}" >/dev/null 2>&1