mirror of
https://github.com/hanzoai/k8s.git
synced 2026-10-02 15:40:05 +00:00
disk: a node agent that keeps every node's disks from filling
At 75% it drops caches (unused images, the journal, stale /tmp and build caches); at 80% it labels the node hanzo.ai/disk=pressure so disk-heavy pods stay off; at 85% it purges rotated logs and the forge's push quarantines, then grows the EBS volume by half (6 h cooldown, 4 TiB ceiling) and the filesystem with it; at 90% it cordons the node. Its own module, standard library only, shipped as ghcr.io/hanzoai/disk. Co-authored-by: Hanzo Dev <[email protected]>
This commit is contained in:
Hanzo Dev
@@ -0,0 +1,14 @@
|
||||
# The disk guard: one static binary in an empty image. Everything it runs on a
|
||||
# node — crictl, journalctl, growpart, resize2fs — is the node's own, reached
|
||||
# through the node's mount namespace, so the image carries no tools.
|
||||
FROM --platform=$BUILDPLATFORM ghcr.io/hanzoai/mirror/golang:1.27.1-alpine@sha256:cf6fca6641884b8433441b2b0652976f975e1d0fdd26d177eaaf8596087f3125 AS build
|
||||
RUN apk add --no-cache ca-certificates
|
||||
WORKDIR /src
|
||||
COPY . .
|
||||
ARG TARGETOS TARGETARCH
|
||||
RUN CGO_ENABLED=0 GOOS=$TARGETOS GOARCH=$TARGETARCH go build -trimpath -ldflags='-s -w' -o /disk .
|
||||
|
||||
FROM scratch
|
||||
COPY --from=build /etc/ssl/certs/ca-certificates.crt /etc/ssl/certs/ca-certificates.crt
|
||||
COPY --from=build /disk /disk
|
||||
ENTRYPOINT ["/disk"]
|
||||
@@ -0,0 +1 @@
|
||||
1.0.0
|
||||
@@ -0,0 +1,265 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"encoding/binary"
|
||||
"encoding/hex"
|
||||
"encoding/xml"
|
||||
"io/fs"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"testing"
|
||||
"time"
|
||||
)
|
||||
|
||||
// The label holds between 75% and 80% whichever way the disk is moving, so a
|
||||
// disk hovering in that band never flips it.
|
||||
func TestPressureHasHysteresis(t *testing.T) {
|
||||
cases := []struct {
|
||||
fill float64
|
||||
now, want bool
|
||||
}{
|
||||
{0.50, false, false}, {0.50, true, false},
|
||||
{0.749, true, false},
|
||||
{0.75, false, false}, {0.75, true, true},
|
||||
{0.79, false, false}, {0.79, true, true},
|
||||
{0.80, false, true}, {0.95, false, true},
|
||||
}
|
||||
for _, c := range cases {
|
||||
if got := pressured(c.fill, c.now); got != c.want {
|
||||
t.Errorf("pressured(%.3f, %v) = %v, want %v", c.fill, c.now, got, c.want)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestCordonHasHysteresis(t *testing.T) {
|
||||
cases := []struct {
|
||||
fill float64
|
||||
now, want bool
|
||||
}{
|
||||
{0.89, false, false}, {0.90, false, true},
|
||||
{0.85, true, true}, {0.80, true, true},
|
||||
{0.799, true, false}, {0.70, false, false},
|
||||
}
|
||||
for _, c := range cases {
|
||||
if got := cordoned(c.fill, c.now); got != c.want {
|
||||
t.Errorf("cordoned(%.3f, %v) = %v, want %v", c.fill, c.now, got, c.want)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func TestGrowthAddsHalfUpToTheCeiling(t *testing.T) {
|
||||
cases := []struct {
|
||||
size, want int
|
||||
ok bool
|
||||
}{
|
||||
{60, 90, true}, {1, 2, true}, {1000, 1500, true}, {1601, 2402, true},
|
||||
{3000, 4096, true}, {4095, 4096, true}, {4096, 4096, false}, {5000, 5000, false},
|
||||
}
|
||||
for _, c := range cases {
|
||||
got, ok := grown(c.size)
|
||||
if got != c.want || ok != c.ok {
|
||||
t.Errorf("grown(%d) = %d, %v; want %d, %v", c.size, got, ok, c.want, c.ok)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// One entry per block device, the last mount at a point wins, and a point that
|
||||
// is not a mount is not watched.
|
||||
func TestMountsWatchesEachDiskOnce(t *testing.T) {
|
||||
info := strings.Join([]string{
|
||||
"24 1 259:1 / / rw,relatime shared:1 - ext4 /dev/root rw,discard",
|
||||
"40 24 0:22 / /proc rw - proc proc rw",
|
||||
"91 24 259:5 / /mnt/hd rw shared:40 - ext4 /dev/nvme1n1 rw",
|
||||
"92 24 259:5 /k3s /var/lib/rancher/k3s/storage rw shared:40 - ext4 /dev/nvme1n1 rw",
|
||||
}, "\n")
|
||||
ms, err := mounts(strings.NewReader(info))
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if len(ms) != 2 || ms[0].Point != "/" || ms[0].Dev != "259:1" || ms[1].Point != "/var/lib/rancher/k3s/storage" || ms[1].Type != "ext4" {
|
||||
t.Fatalf("mounts = %+v", ms)
|
||||
}
|
||||
// The storage directory on the root disk is the root disk.
|
||||
ms, _ = mounts(strings.NewReader("24 1 259:1 / / rw - ext4 /dev/root rw\n25 24 259:1 /x /var/lib/rancher/k3s/storage rw - ext4 /dev/root rw\n"))
|
||||
if len(ms) != 1 {
|
||||
t.Fatalf("a bind of the root disk counted twice: %+v", ms)
|
||||
}
|
||||
if _, err := mounts(strings.NewReader("40 24 0:22 / /proc rw - proc proc rw\n")); err == nil {
|
||||
t.Fatal("no watched mount is an error, not an empty answer")
|
||||
}
|
||||
}
|
||||
|
||||
// A partition resolves to its disk, the room after it, and the EBS volume named
|
||||
// by the disk's NVMe serial.
|
||||
func TestDeviceReadsSysfs(t *testing.T) {
|
||||
root := t.TempDir()
|
||||
disk := filepath.Join(root, "sys/devices/pci0000:00/nvme/nvme0/nvme0n1")
|
||||
part := filepath.Join(disk, "nvme0n1p1")
|
||||
write(t, filepath.Join(disk, "size"), "2097152000") // 1000 GiB
|
||||
write(t, filepath.Join(part, "size"), "125829120") // 60 GiB
|
||||
write(t, filepath.Join(part, "start"), "2048")
|
||||
write(t, filepath.Join(part, "partition"), "1")
|
||||
write(t, filepath.Join(root, "sys/block/nvme0n1/device/serial"), "vol0c45cf5a13bfd27e4 \n")
|
||||
link(t, part, filepath.Join(root, "sys/dev/block/259:1"))
|
||||
|
||||
b, err := device(root, "259:1")
|
||||
if err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if b.Name != "nvme0n1p1" || b.Disk != "nvme0n1" || b.Part != 1 || b.Bytes != 60<<30 || b.Volume != "vol-0c45cf5a13bfd27e4" {
|
||||
t.Fatalf("device = %+v", b)
|
||||
}
|
||||
if want := uint64(2097152000-125829120-2048) * 512; b.Room != want {
|
||||
t.Fatalf("room = %d, want %d", b.Room, want)
|
||||
}
|
||||
|
||||
whole := filepath.Join(root, "sys/devices/pci0000:00/nvme/nvme1/nvme1n1")
|
||||
write(t, filepath.Join(whole, "size"), "3355443200")
|
||||
link(t, whole, filepath.Join(root, "sys/dev/block/259:5"))
|
||||
if b, err = device(root, "259:5"); err != nil || b.Part != 0 || b.Disk != "nvme1n1" || b.Room != 0 || b.Volume != "" {
|
||||
t.Fatalf("whole disk = %+v, %v", b, err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestSuperblockSize(t *testing.T) {
|
||||
ext := make([]byte, 2048)
|
||||
s := ext[1024:]
|
||||
binary.LittleEndian.PutUint16(s[0x38:], 0xEF53)
|
||||
binary.LittleEndian.PutUint32(s[0x18:], 2) // 4 KiB blocks
|
||||
binary.LittleEndian.PutUint32(s[0x04:], 0x10)
|
||||
binary.LittleEndian.PutUint32(s[0x60:], 0x80) // 64bit
|
||||
binary.LittleEndian.PutUint32(s[0x150:], 1)
|
||||
if got, err := superblock(ext, "ext4"); err != nil || got != (1<<32|0x10)*4096 {
|
||||
t.Fatalf("ext4 = %d, %v", got, err)
|
||||
}
|
||||
xfs := make([]byte, 2048)
|
||||
copy(xfs, "XFSB")
|
||||
binary.BigEndian.PutUint32(xfs[4:], 4096)
|
||||
binary.BigEndian.PutUint64(xfs[8:], 1000)
|
||||
if got, err := superblock(xfs, "xfs"); err != nil || got != 4096000 {
|
||||
t.Fatalf("xfs = %d, %v", got, err)
|
||||
}
|
||||
if _, err := superblock(make([]byte, 2048), "ext4"); err == nil {
|
||||
t.Fatal("a missing magic read as a size")
|
||||
}
|
||||
}
|
||||
|
||||
func TestRotatedLogs(t *testing.T) {
|
||||
for name, want := range map[string]bool{
|
||||
"syslog.1": true, "kern.log.2.gz": true, "0.log.20261002-120102.gz": true, "dpkg.log.old": true,
|
||||
"syslog": false, "kern.log": false, "0.log.20261002-120102": false, "cloud-init.log": false,
|
||||
} {
|
||||
if got := rotated("", entry{name}); got != want {
|
||||
t.Errorf("rotated(%q) = %v, want %v", name, got, want)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// A build cache is found by its tag and kept while anything inside is fresh.
|
||||
func TestCachesAreTaggedAndIdle(t *testing.T) {
|
||||
root := t.TempDir()
|
||||
old := time.Now().Add(-48 * time.Hour)
|
||||
stale := filepath.Join(root, "home/ubuntu/engine/target")
|
||||
fresh := filepath.Join(root, "home/ubuntu/ml/target")
|
||||
gobuild := filepath.Join(root, "root/.cache/go-build")
|
||||
plain := filepath.Join(root, "home/ubuntu/src")
|
||||
for _, d := range []string{stale, fresh} {
|
||||
write(t, filepath.Join(d, "CACHEDIR.TAG"), "Signature: 8a477f597d28d172789f06886806bc55\n")
|
||||
write(t, filepath.Join(d, "debug/x"), "x")
|
||||
}
|
||||
write(t, filepath.Join(gobuild, "00/a"), "x")
|
||||
write(t, filepath.Join(plain, "main.go"), "package main")
|
||||
for _, d := range []string{stale, gobuild, plain} {
|
||||
age(t, d, old)
|
||||
}
|
||||
age(t, filepath.Join(fresh, "CACHEDIR.TAG"), old)
|
||||
|
||||
got := caches(root)
|
||||
if len(got) != 2 || got[0] != gobuild && got[1] != gobuild || !contains(got, stale) {
|
||||
t.Fatalf("caches = %v, want %s and %s", got, stale, gobuild)
|
||||
}
|
||||
}
|
||||
|
||||
func TestQuarantinesAreIdleOnly(t *testing.T) {
|
||||
root := t.TempDir()
|
||||
repos := filepath.Join(root, "var/lib/rancher/k3s/storage/pvc-1_hanzo_hanzo-git-data/git/data/repositories")
|
||||
idle := filepath.Join(repos, "hanzoai/team.git/objects/tmp_objdir-incoming-a")
|
||||
busy := filepath.Join(repos, "hanzoai/ui.git/objects/tmp_objdir-incoming-b")
|
||||
write(t, filepath.Join(idle, "pack/p"), "x")
|
||||
write(t, filepath.Join(busy, "pack/p"), "x")
|
||||
age(t, idle, time.Now().Add(-time.Hour))
|
||||
age(t, filepath.Join(busy, "pack"), time.Now().Add(-time.Hour))
|
||||
got := quarantines(root, 10*time.Minute)
|
||||
if len(got) != 1 || got[0] != idle {
|
||||
t.Fatalf("quarantines = %v, want only %s", got, idle)
|
||||
}
|
||||
}
|
||||
|
||||
// AWS's published derivation example (Signature Version 4, "deriving a signing key").
|
||||
func TestSigningKey(t *testing.T) {
|
||||
got := hex.EncodeToString(key("wJalrXUtnFEMI/K7MDENG+bPxRfiCYEXAMPLEKEY", "20120215", "us-east-1", "iam"))
|
||||
if got != "f4780e2d9f65fa895f9c67b32ce1baf0b0d8a43505a000a1a9e090d414db404d" {
|
||||
t.Fatalf("signing key = %s", got)
|
||||
}
|
||||
}
|
||||
|
||||
func TestModificationsDecode(t *testing.T) {
|
||||
var out struct {
|
||||
Mods []struct {
|
||||
State string `xml:"modificationState"`
|
||||
Start time.Time `xml:"startTime"`
|
||||
} `xml:"volumeModificationSet>item"`
|
||||
}
|
||||
body := `<DescribeVolumesModificationsResponse xmlns="http://ec2.amazonaws.com/doc/2016-11-15/"><volumeModificationSet><item><volumeId>vol-1</volumeId><modificationState>optimizing</modificationState><startTime>2026-10-02T09:10:11.000Z</startTime></item></volumeModificationSet></DescribeVolumesModificationsResponse>`
|
||||
if err := xml.Unmarshal([]byte(body), &out); err != nil || len(out.Mods) != 1 || out.Mods[0].State != "optimizing" || out.Mods[0].Start.Hour() != 9 {
|
||||
t.Fatalf("decoded %+v, %v", out, err)
|
||||
}
|
||||
}
|
||||
|
||||
type entry struct{ name string }
|
||||
|
||||
func (e entry) Name() string { return e.name }
|
||||
func (e entry) IsDir() bool { return false }
|
||||
func (e entry) Type() fs.FileMode { return 0 }
|
||||
func (e entry) Info() (fs.FileInfo, error) { return nil, nil }
|
||||
|
||||
func write(t *testing.T, path, s string) {
|
||||
t.Helper()
|
||||
if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := os.WriteFile(path, []byte(s), 0o644); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
|
||||
func link(t *testing.T, target, at string) {
|
||||
t.Helper()
|
||||
if err := os.MkdirAll(filepath.Dir(at), 0o755); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
if err := os.Symlink(target, at); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
|
||||
// age sets the time of path and everything under it.
|
||||
func age(t *testing.T, path string, when time.Time) {
|
||||
t.Helper()
|
||||
filepath.WalkDir(path, func(p string, d fs.DirEntry, err error) error {
|
||||
if err == nil {
|
||||
os.Chtimes(p, when, when)
|
||||
}
|
||||
return nil
|
||||
})
|
||||
}
|
||||
|
||||
func contains(s []string, v string) bool {
|
||||
for _, x := range s {
|
||||
if x == v {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
+206
@@ -0,0 +1,206 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/hmac"
|
||||
"crypto/sha256"
|
||||
"encoding/hex"
|
||||
"encoding/json"
|
||||
"encoding/xml"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"net/http"
|
||||
"net/url"
|
||||
"strconv"
|
||||
"strings"
|
||||
"time"
|
||||
)
|
||||
|
||||
// ebs grows the node's own EBS volumes with the node's instance role, whose
|
||||
// policy (infra/aws/main.tf, "disk") allows ec2:ModifyVolume only on volumes
|
||||
// tagged hanzo.ai/disk=grow. Credentials come from the instance metadata
|
||||
// service, which answers only on the node's own network (hop limit 1), which is
|
||||
// why this pod runs on the host network.
|
||||
type ebs struct {
|
||||
c *http.Client
|
||||
region string
|
||||
}
|
||||
|
||||
const imds = "http://169.254.169.254"
|
||||
|
||||
func (e *ebs) meta(ctx context.Context, path string) (string, error) {
|
||||
req, _ := http.NewRequestWithContext(ctx, http.MethodPut, imds+"/latest/api/token", nil)
|
||||
req.Header.Set("X-aws-ec2-metadata-token-ttl-seconds", "300")
|
||||
resp, err := e.c.Do(req)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
tok, _ := io.ReadAll(io.LimitReader(resp.Body, 4096))
|
||||
resp.Body.Close()
|
||||
if resp.StatusCode != http.StatusOK {
|
||||
return "", fmt.Errorf("imds token: %s", resp.Status)
|
||||
}
|
||||
req, _ = http.NewRequestWithContext(ctx, http.MethodGet, imds+path, nil)
|
||||
req.Header.Set("X-aws-ec2-metadata-token", string(tok))
|
||||
resp, err = e.c.Do(req)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
b, _ := io.ReadAll(io.LimitReader(resp.Body, 1<<16))
|
||||
if resp.StatusCode != http.StatusOK {
|
||||
return "", fmt.Errorf("imds %s: %s", path, resp.Status)
|
||||
}
|
||||
return strings.TrimSpace(string(b)), nil
|
||||
}
|
||||
|
||||
type creds struct {
|
||||
AccessKeyId, SecretAccessKey, Token string
|
||||
}
|
||||
|
||||
func (e *ebs) creds(ctx context.Context) (creds, error) {
|
||||
role, err := e.meta(ctx, "/latest/meta-data/iam/security-credentials/")
|
||||
if err != nil {
|
||||
return creds{}, err
|
||||
}
|
||||
raw, err := e.meta(ctx, "/latest/meta-data/iam/security-credentials/"+strings.Fields(role)[0])
|
||||
if err != nil {
|
||||
return creds{}, err
|
||||
}
|
||||
var c creds
|
||||
if err := json.Unmarshal([]byte(raw), &c); err != nil {
|
||||
return creds{}, err
|
||||
}
|
||||
return c, nil
|
||||
}
|
||||
|
||||
// call makes one EC2 Query API call, signed with Signature Version 4.
|
||||
func (e *ebs) call(ctx context.Context, params url.Values, out any) error {
|
||||
c, err := e.creds(ctx)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
params.Set("Version", "2016-11-15")
|
||||
body := params.Encode()
|
||||
hostname := "ec2." + e.region + ".amazonaws.com"
|
||||
req, _ := http.NewRequestWithContext(ctx, http.MethodPost, "https://"+hostname+"/", strings.NewReader(body))
|
||||
req.Header.Set("Content-Type", "application/x-www-form-urlencoded; charset=utf-8")
|
||||
sign(req, body, hostname, e.region, "ec2", c, time.Now())
|
||||
resp, err := e.c.Do(req)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
data, _ := io.ReadAll(io.LimitReader(resp.Body, 1<<20))
|
||||
if resp.StatusCode != http.StatusOK {
|
||||
var fault struct {
|
||||
Errors []struct {
|
||||
Code string `xml:"Code"`
|
||||
Message string `xml:"Message"`
|
||||
} `xml:"Errors>Error"`
|
||||
}
|
||||
if xml.Unmarshal(data, &fault) == nil && len(fault.Errors) > 0 {
|
||||
return &awsError{Code: fault.Errors[0].Code, Message: fault.Errors[0].Message}
|
||||
}
|
||||
return fmt.Errorf("%s: %s", params.Get("Action"), resp.Status)
|
||||
}
|
||||
return xml.Unmarshal(data, out)
|
||||
}
|
||||
|
||||
type awsError struct{ Code, Message string }
|
||||
|
||||
func (e *awsError) Error() string { return e.Code + ": " + e.Message }
|
||||
|
||||
// sign adds a Signature Version 4 Authorization header to req.
|
||||
func sign(req *http.Request, body, hostname, region, service string, c creds, now time.Time) {
|
||||
stamp := now.UTC().Format("20060102T150405Z")
|
||||
date := stamp[:8]
|
||||
req.Header.Set("X-Amz-Date", stamp)
|
||||
headers := []string{"content-type", "host", "x-amz-date"}
|
||||
values := map[string]string{"content-type": req.Header.Get("Content-Type"), "host": hostname, "x-amz-date": stamp}
|
||||
if c.Token != "" {
|
||||
req.Header.Set("X-Amz-Security-Token", c.Token)
|
||||
headers = append(headers, "x-amz-security-token")
|
||||
values["x-amz-security-token"] = c.Token
|
||||
}
|
||||
var canon strings.Builder
|
||||
for _, h := range headers {
|
||||
canon.WriteString(h + ":" + values[h] + "\n")
|
||||
}
|
||||
signed := strings.Join(headers, ";")
|
||||
request := strings.Join([]string{req.Method, "/", "", canon.String(), signed, hexSum(body)}, "\n")
|
||||
scope := date + "/" + region + "/" + service + "/aws4_request"
|
||||
toSign := strings.Join([]string{"AWS4-HMAC-SHA256", stamp, scope, hexSum(request)}, "\n")
|
||||
sig := hex.EncodeToString(mac(key(c.SecretAccessKey, date, region, service), toSign))
|
||||
req.Header.Set("Authorization", "AWS4-HMAC-SHA256 Credential="+c.AccessKeyId+"/"+scope+", SignedHeaders="+signed+", Signature="+sig)
|
||||
}
|
||||
|
||||
func key(secret, date, region, service string) []byte {
|
||||
k := mac([]byte("AWS4"+secret), date)
|
||||
k = mac(k, region)
|
||||
k = mac(k, service)
|
||||
return mac(k, "aws4_request")
|
||||
}
|
||||
|
||||
func mac(k []byte, s string) []byte {
|
||||
h := hmac.New(sha256.New, k)
|
||||
h.Write([]byte(s))
|
||||
return h.Sum(nil)
|
||||
}
|
||||
|
||||
func hexSum(s string) string {
|
||||
h := sha256.Sum256([]byte(s))
|
||||
return hex.EncodeToString(h[:])
|
||||
}
|
||||
|
||||
// size is the volume's size in GiB.
|
||||
func (e *ebs) size(ctx context.Context, vol string) (int, error) {
|
||||
var out struct {
|
||||
Volumes []struct {
|
||||
Size string `xml:"size"`
|
||||
} `xml:"volumeSet>item"`
|
||||
}
|
||||
if err := e.call(ctx, url.Values{"Action": {"DescribeVolumes"}, "VolumeId.1": {vol}}, &out); err != nil {
|
||||
return 0, err
|
||||
}
|
||||
if len(out.Volumes) != 1 {
|
||||
return 0, fmt.Errorf("%s: described %d volumes", vol, len(out.Volumes))
|
||||
}
|
||||
return strconv.Atoi(out.Volumes[0].Size)
|
||||
}
|
||||
|
||||
// changed is when the volume's latest modification started, and its state;
|
||||
// zero for a volume never modified.
|
||||
func (e *ebs) changed(ctx context.Context, vol string) (time.Time, string, error) {
|
||||
var out struct {
|
||||
Mods []struct {
|
||||
State string `xml:"modificationState"`
|
||||
Start time.Time `xml:"startTime"`
|
||||
} `xml:"volumeModificationSet>item"`
|
||||
}
|
||||
err := e.call(ctx, url.Values{"Action": {"DescribeVolumesModifications"}, "VolumeId.1": {vol}}, &out)
|
||||
var ae *awsError
|
||||
if errors.As(err, &ae) && ae.Code == "InvalidVolumeModification.NotFound" {
|
||||
return time.Time{}, "", nil
|
||||
}
|
||||
if err != nil {
|
||||
return time.Time{}, "", err
|
||||
}
|
||||
var t time.Time
|
||||
var state string
|
||||
for _, m := range out.Mods {
|
||||
if m.Start.After(t) {
|
||||
t, state = m.Start, m.State
|
||||
}
|
||||
}
|
||||
return t, state, nil
|
||||
}
|
||||
|
||||
func (e *ebs) grow(ctx context.Context, vol string, size int) error {
|
||||
var out struct{}
|
||||
return e.call(ctx, url.Values{"Action": {"ModifyVolume"}, "VolumeId": {vol}, "Size": {strconv.Itoa(size)}}, &out)
|
||||
}
|
||||
|
||||
// cooldown is how long AWS makes a volume wait between modifications.
|
||||
const cooldown = 6 * time.Hour
|
||||
+192
@@ -0,0 +1,192 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"bufio"
|
||||
"encoding/binary"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strconv"
|
||||
"strings"
|
||||
"syscall"
|
||||
)
|
||||
|
||||
// host is the node's root as this process sees it. With hostPID, PID 1 is the
|
||||
// node's init, and its root is the node's filesystem with every mount on it —
|
||||
// no hostPath volume, and nothing of the node mounted into this container.
|
||||
const host = "/proc/1/root"
|
||||
|
||||
// watched are the mount points this program keeps from filling: the root, which
|
||||
// holds images, logs and the kubelet, and the disk k3s keeps local volumes on
|
||||
// where a node has one.
|
||||
var watched = []string{"/", "/var/lib/rancher/k3s/storage"}
|
||||
|
||||
// mount is one watched filesystem.
|
||||
type mount struct {
|
||||
Point string // where it is mounted on the node
|
||||
Dev string // major:minor of its block device
|
||||
Type string // ext4, xfs
|
||||
Fill float64
|
||||
Size uint64 // bytes the filesystem reports, for logs
|
||||
Avail uint64
|
||||
}
|
||||
|
||||
// mounts reads the node's mountinfo (/proc/1/mountinfo) and returns the
|
||||
// watched mount points that exist, one per block device: a bind of the same
|
||||
// device under a second name is the same disk and is counted once.
|
||||
func mounts(mountinfo io.Reader) ([]mount, error) {
|
||||
found := map[string]mount{}
|
||||
sc := bufio.NewScanner(mountinfo)
|
||||
sc.Buffer(make([]byte, 64<<10), 1<<20)
|
||||
for sc.Scan() {
|
||||
f := strings.Fields(sc.Text())
|
||||
sep := -1
|
||||
for i, s := range f {
|
||||
if s == "-" {
|
||||
sep = i
|
||||
break
|
||||
}
|
||||
}
|
||||
if sep < 6 || sep+1 >= len(f) {
|
||||
continue
|
||||
}
|
||||
point := f[4]
|
||||
for _, w := range watched {
|
||||
if point == w {
|
||||
// Later lines mount over earlier ones; the last is what a path resolves to.
|
||||
found[w] = mount{Point: w, Dev: f[2], Type: f[sep+1]}
|
||||
}
|
||||
}
|
||||
}
|
||||
if err := sc.Err(); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
var out []mount
|
||||
seen := map[string]bool{}
|
||||
for _, w := range watched {
|
||||
m, ok := found[w]
|
||||
if !ok || seen[m.Dev] {
|
||||
continue
|
||||
}
|
||||
seen[m.Dev] = true
|
||||
out = append(out, m)
|
||||
}
|
||||
if len(out) == 0 {
|
||||
return nil, errors.New("no watched filesystem in mountinfo")
|
||||
}
|
||||
return out, nil
|
||||
}
|
||||
|
||||
// measure fills in how full m is, the way df reports it: used over what an
|
||||
// unprivileged writer could ever use (used + available), so the reserved blocks
|
||||
// count as full.
|
||||
func measure(m *mount) error {
|
||||
var st syscall.Statfs_t
|
||||
if err := syscall.Statfs(host+m.Point, &st); err != nil {
|
||||
return err
|
||||
}
|
||||
bs := uint64(st.Bsize)
|
||||
used := (st.Blocks - st.Bfree) * bs
|
||||
m.Avail = st.Bavail * bs
|
||||
m.Size = st.Blocks * bs
|
||||
if used+m.Avail == 0 {
|
||||
return fmt.Errorf("%s reports no blocks", m.Point)
|
||||
}
|
||||
m.Fill = float64(used) / float64(used+m.Avail)
|
||||
return nil
|
||||
}
|
||||
|
||||
// block is the device under a filesystem, read from sysfs.
|
||||
type block struct {
|
||||
Name string // nvme0n1p1, the node /dev/<Name>
|
||||
Disk string // nvme0n1, the whole disk the volume is
|
||||
Part int // partition number, 0 for a whole disk
|
||||
Bytes uint64 // size of Name
|
||||
Room uint64 // bytes of Disk after the end of Name
|
||||
Volume string // EBS volume id, from the NVMe serial; empty off EBS
|
||||
}
|
||||
|
||||
// device resolves major:minor through sysfs (under root, normally host).
|
||||
func device(root, dev string) (block, error) {
|
||||
link := filepath.Join(root, "sys/dev/block", dev)
|
||||
target, err := filepath.EvalSymlinks(link)
|
||||
if err != nil {
|
||||
return block{}, err
|
||||
}
|
||||
b := block{Name: filepath.Base(target), Disk: filepath.Base(target)}
|
||||
sectors, err := readUint(filepath.Join(target, "size"))
|
||||
if err != nil {
|
||||
return block{}, err
|
||||
}
|
||||
b.Bytes = sectors * 512
|
||||
if p, err := readUint(filepath.Join(target, "partition")); err == nil {
|
||||
b.Part = int(p)
|
||||
b.Disk = filepath.Base(filepath.Dir(target))
|
||||
start, err := readUint(filepath.Join(target, "start"))
|
||||
if err != nil {
|
||||
return block{}, err
|
||||
}
|
||||
disk, err := readUint(filepath.Join(filepath.Dir(target), "size"))
|
||||
if err != nil {
|
||||
return block{}, err
|
||||
}
|
||||
if end := start + sectors; disk > end {
|
||||
b.Room = (disk - end) * 512
|
||||
}
|
||||
}
|
||||
// An EBS volume on a Nitro instance is an NVMe namespace whose serial is the
|
||||
// volume id without its dash: vol0123… for vol-0123….
|
||||
if s, err := os.ReadFile(filepath.Join(root, "sys/block", b.Disk, "device/serial")); err == nil {
|
||||
if v := strings.TrimSpace(string(s)); strings.HasPrefix(v, "vol") && !strings.HasPrefix(v, "vol-") {
|
||||
b.Volume = "vol-" + strings.TrimPrefix(v, "vol")
|
||||
}
|
||||
}
|
||||
return b, nil
|
||||
}
|
||||
|
||||
func readUint(path string) (uint64, error) {
|
||||
s, err := os.ReadFile(path)
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
return strconv.ParseUint(strings.TrimSpace(string(s)), 10, 64)
|
||||
}
|
||||
|
||||
// fsBytes is the size the filesystem on dev records in its own superblock —
|
||||
// not statfs, which subtracts ext4's metadata and so always reads a few percent
|
||||
// under the device. Equal to the device once the filesystem fills it.
|
||||
func fsBytes(dev, kind string) (uint64, error) {
|
||||
f, err := os.Open(dev)
|
||||
if err != nil {
|
||||
return 0, err
|
||||
}
|
||||
defer f.Close()
|
||||
sb := make([]byte, 2048)
|
||||
if _, err := io.ReadFull(f, sb); err != nil {
|
||||
return 0, err
|
||||
}
|
||||
return superblock(sb, kind)
|
||||
}
|
||||
|
||||
func superblock(sb []byte, kind string) (uint64, error) {
|
||||
switch kind {
|
||||
case "ext4", "ext3", "ext2":
|
||||
s := sb[1024:]
|
||||
if binary.LittleEndian.Uint16(s[0x38:]) != 0xEF53 {
|
||||
return 0, errors.New("no ext superblock")
|
||||
}
|
||||
blocks := uint64(binary.LittleEndian.Uint32(s[0x04:]))
|
||||
if binary.LittleEndian.Uint32(s[0x60:])&0x80 != 0 { // INCOMPAT_64BIT
|
||||
blocks |= uint64(binary.LittleEndian.Uint32(s[0x150:])) << 32
|
||||
}
|
||||
return blocks << (10 + binary.LittleEndian.Uint32(s[0x18:])), nil
|
||||
case "xfs":
|
||||
if string(sb[:4]) != "XFSB" {
|
||||
return 0, errors.New("no xfs superblock")
|
||||
}
|
||||
return binary.BigEndian.Uint64(sb[8:]) * uint64(binary.BigEndian.Uint32(sb[4:])), nil
|
||||
}
|
||||
return 0, fmt.Errorf("cannot read the size of a %s filesystem", kind)
|
||||
}
|
||||
@@ -0,0 +1,3 @@
|
||||
module github.com/hanzoai/k8s/disk
|
||||
|
||||
go 1.27.1
|
||||
+189
@@ -0,0 +1,189 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"io/fs"
|
||||
"log/slog"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"strings"
|
||||
"syscall"
|
||||
"time"
|
||||
)
|
||||
|
||||
// evict drops what the node keeps only to save a download or a rebuild.
|
||||
func evict(ctx context.Context, log *slog.Logger) {
|
||||
step(ctx, log, "images", "/usr/local/bin/crictl", "rmi", "--prune")
|
||||
step(ctx, log, "journal", "/usr/bin/journalctl", "--vacuum-size=256M")
|
||||
log.Info("evict tmp", "bytes", sweep(host+"/tmp", day, plainFiles))
|
||||
for _, c := range caches(host) {
|
||||
log.Info("evict cache", "path", strings.TrimPrefix(c, host), "bytes", remove(c))
|
||||
}
|
||||
}
|
||||
|
||||
// purge is evict and then what is regenerable but costs more to lose: a shorter
|
||||
// journal, rotated logs, and push quarantines the forge left behind.
|
||||
func purge(ctx context.Context, log *slog.Logger) {
|
||||
evict(ctx, log)
|
||||
step(ctx, log, "journal", "/usr/bin/journalctl", "--vacuum-size=64M")
|
||||
log.Info("purge logs", "bytes", sweep(host+"/var/log", 0, rotated))
|
||||
for _, q := range quarantines(host, 10*time.Minute) {
|
||||
log.Info("purge quarantine", "path", strings.TrimPrefix(q, host), "bytes", remove(q))
|
||||
}
|
||||
}
|
||||
|
||||
func step(ctx context.Context, log *slog.Logger, what string, argv ...string) {
|
||||
ctx, cancel := context.WithTimeout(ctx, 10*time.Minute)
|
||||
defer cancel()
|
||||
out, err := run(ctx, argv...)
|
||||
if err != nil {
|
||||
log.Warn("evict "+what, "err", err, "out", out)
|
||||
return
|
||||
}
|
||||
log.Info("evict "+what, "out", last(out))
|
||||
}
|
||||
|
||||
func last(s string) string {
|
||||
if i := strings.LastIndexByte(s, '\n'); i >= 0 {
|
||||
return s[i+1:]
|
||||
}
|
||||
return s
|
||||
}
|
||||
|
||||
const day = 24 * time.Hour
|
||||
|
||||
func plainFiles(path string, d fs.DirEntry) bool { return d.Type().IsRegular() }
|
||||
|
||||
// rotated is a log a rotation already closed: syslog.1, kern.log.2.gz.
|
||||
func rotated(path string, d fs.DirEntry) bool {
|
||||
if !d.Type().IsRegular() {
|
||||
return false
|
||||
}
|
||||
n := d.Name()
|
||||
if strings.HasSuffix(n, ".gz") || strings.HasSuffix(n, ".xz") || strings.HasSuffix(n, ".old") {
|
||||
return true
|
||||
}
|
||||
ext := filepath.Ext(n)
|
||||
return len(ext) > 1 && strings.Trim(ext[1:], "0123456789") == ""
|
||||
}
|
||||
|
||||
// sweep removes, under root and on root's own filesystem, every file that match
|
||||
// accepts and that was last written longer than age ago, and returns the bytes.
|
||||
func sweep(root string, age time.Duration, match func(string, fs.DirEntry) bool) uint64 {
|
||||
dev, ok := devOf(root)
|
||||
if !ok {
|
||||
return 0
|
||||
}
|
||||
cut := time.Now().Add(-age)
|
||||
var freed uint64
|
||||
filepath.WalkDir(root, func(p string, d fs.DirEntry, err error) error {
|
||||
if err != nil {
|
||||
return nil
|
||||
}
|
||||
if d.IsDir() {
|
||||
if dd, ok := devOf(p); ok && dd != dev {
|
||||
return filepath.SkipDir // another filesystem mounted inside
|
||||
}
|
||||
return nil
|
||||
}
|
||||
if !match(p, d) {
|
||||
return nil
|
||||
}
|
||||
info, err := d.Info()
|
||||
if err != nil || info.ModTime().After(cut) {
|
||||
return nil
|
||||
}
|
||||
if os.Remove(p) == nil {
|
||||
freed += uint64(info.Size())
|
||||
}
|
||||
return nil
|
||||
})
|
||||
return freed
|
||||
}
|
||||
|
||||
func devOf(p string) (uint64, bool) {
|
||||
var st syscall.Stat_t
|
||||
if syscall.Lstat(p, &st) != nil {
|
||||
return 0, false
|
||||
}
|
||||
return uint64(st.Dev), true
|
||||
}
|
||||
|
||||
// caches are the build caches under the node's home directories that nothing has
|
||||
// written to for a day. A cache says so itself: cargo's target directories carry
|
||||
// CACHEDIR.TAG, as does any tool following that convention, and Go's is named
|
||||
// go-build. Age is the NEWEST file inside, never the directory's own mtime, which
|
||||
// changes only when a direct child is added or removed.
|
||||
func caches(root string) []string {
|
||||
homes, _ := filepath.Glob(root + "/home/*")
|
||||
homes = append(homes, root+"/root")
|
||||
var out []string
|
||||
cut := time.Now().Add(-day)
|
||||
for _, h := range homes {
|
||||
base := strings.Count(h, string(os.PathSeparator))
|
||||
filepath.WalkDir(h, func(p string, d fs.DirEntry, err error) error {
|
||||
if err != nil || !d.IsDir() || d.Type()&fs.ModeSymlink != 0 {
|
||||
return nil
|
||||
}
|
||||
if strings.Count(p, string(os.PathSeparator))-base > 4 {
|
||||
return filepath.SkipDir
|
||||
}
|
||||
_, tagged := os.Stat(filepath.Join(p, "CACHEDIR.TAG"))
|
||||
if tagged != nil && d.Name() != "go-build" {
|
||||
return nil
|
||||
}
|
||||
if newest(p).Before(cut) {
|
||||
out = append(out, p)
|
||||
}
|
||||
return filepath.SkipDir
|
||||
})
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// quarantines are the forge's push quarantines older than age: git writes a
|
||||
// push's objects into objects/tmp_objdir-incoming-* and removes the directory
|
||||
// when the push ends, but not when the connection is cut.
|
||||
func quarantines(root string, age time.Duration) []string {
|
||||
dirs, _ := filepath.Glob(root + "/var/lib/rancher/k3s/storage/*hanzo-git-data*/git/data/repositories/*/*/objects/tmp_objdir-incoming-*")
|
||||
cut := time.Now().Add(-age)
|
||||
var out []string
|
||||
for _, d := range dirs {
|
||||
if newest(d).Before(cut) {
|
||||
out = append(out, d)
|
||||
}
|
||||
}
|
||||
return out
|
||||
}
|
||||
|
||||
// newest is the latest modification time of anything under dir, dir included.
|
||||
func newest(dir string) time.Time {
|
||||
var t time.Time
|
||||
filepath.WalkDir(dir, func(p string, d fs.DirEntry, err error) error {
|
||||
if err != nil {
|
||||
return nil
|
||||
}
|
||||
if info, err := d.Info(); err == nil && info.ModTime().After(t) {
|
||||
t = info.ModTime()
|
||||
}
|
||||
return nil
|
||||
})
|
||||
return t
|
||||
}
|
||||
|
||||
// remove deletes path and returns the bytes it held.
|
||||
func remove(path string) uint64 {
|
||||
var n uint64
|
||||
filepath.WalkDir(path, func(p string, d fs.DirEntry, err error) error {
|
||||
if err == nil && d.Type().IsRegular() {
|
||||
if info, err := d.Info(); err == nil {
|
||||
n += uint64(info.Size())
|
||||
}
|
||||
}
|
||||
return nil
|
||||
})
|
||||
if os.RemoveAll(path) != nil {
|
||||
return 0
|
||||
}
|
||||
return n
|
||||
}
|
||||
@@ -0,0 +1,68 @@
|
||||
package main
|
||||
|
||||
import "math"
|
||||
|
||||
// The ladder: what a node does as a watched filesystem fills. A node's disks are
|
||||
// a cache of things that live elsewhere, so every rung below the last one gives
|
||||
// space back or stops taking more; only the last one asks a person.
|
||||
const (
|
||||
evictAt = 0.75 // drop caches: unused images, journal, stale /tmp and build caches
|
||||
admitAt = 0.80 // stop admitting disk-heavy pods: label the node hanzo.ai/disk=pressure
|
||||
purgeAt = 0.85 // emergency purge, then grow the volume
|
||||
cordonAt = 0.90 // cordon the node; o11y pages on the same reading
|
||||
)
|
||||
|
||||
// Label is the node label disk-heavy pods carry a NotIn affinity against.
|
||||
const (
|
||||
Label = "hanzo.ai/disk"
|
||||
Pressure = "pressure"
|
||||
// Cordoned marks a cordon this program set, so it never lifts one a person
|
||||
// set for another reason.
|
||||
Cordoned = "hanzo.ai/disk-cordon"
|
||||
)
|
||||
|
||||
// pressured is whether the node should carry Label=Pressure at fill f, given
|
||||
// whether it carries it now. It is set at admitAt and cleared below evictAt; in
|
||||
// between the node keeps what it has, so a disk hovering at 78% does not flip
|
||||
// the label every minute.
|
||||
func pressured(f float64, now bool) bool {
|
||||
switch {
|
||||
case f >= admitAt:
|
||||
return true
|
||||
case f < evictAt:
|
||||
return false
|
||||
}
|
||||
return now
|
||||
}
|
||||
|
||||
// cordoned is whether this program's cordon should hold at fill f, given whether
|
||||
// it holds now. Set at cordonAt, lifted below admitAt.
|
||||
func cordoned(f float64, now bool) bool {
|
||||
switch {
|
||||
case f >= cordonAt:
|
||||
return true
|
||||
case f < admitAt:
|
||||
return false
|
||||
}
|
||||
return now
|
||||
}
|
||||
|
||||
// Growth bounds. AWS allows one modification per volume every six hours; a grow
|
||||
// adds half the volume, and none is taken past maxGiB.
|
||||
const (
|
||||
maxGiB = 4096
|
||||
growRate = 1.5
|
||||
)
|
||||
|
||||
// grown is the size, in GiB, a volume of size GiB grows to, and whether there is
|
||||
// any room left to grow at all.
|
||||
func grown(size int) (int, bool) {
|
||||
if size >= maxGiB {
|
||||
return size, false
|
||||
}
|
||||
n := int(math.Ceil(float64(size) * growRate))
|
||||
if n > maxGiB {
|
||||
n = maxGiB
|
||||
}
|
||||
return n, n > size
|
||||
}
|
||||
+256
@@ -0,0 +1,256 @@
|
||||
// Command disk keeps a node's disks from filling. It runs on every node of the
|
||||
// cluster (infra/aws/k8s/disk.yaml) and climbs one ladder per watched
|
||||
// filesystem, every minute:
|
||||
//
|
||||
// 75% evict caches: unused images, the journal past 256M, /tmp files and build
|
||||
// caches nothing has touched for a day
|
||||
// 80% label the node hanzo.ai/disk=pressure, which every disk-heavy pod (builds,
|
||||
// CI runners, sandboxes) carries a NotIn affinity against; cleared below 75%
|
||||
// 85% purge: the journal past 64M, rotated logs, the forge's push quarantines
|
||||
// older than ten minutes; then, still at 85%, grow the EBS volume by half
|
||||
// (to at most 4 TiB, once per six hours) and the filesystem with it
|
||||
// 90% cordon the node; lifted below 80%. o11y pages on the same reading
|
||||
// (NodeDiskCritical)
|
||||
//
|
||||
// A node's disks are a cache of what lives in object storage, so none of this
|
||||
// touches a volume's data: it removes what the node can download or rebuild,
|
||||
// stops taking more, and buys room.
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"log/slog"
|
||||
"net/http"
|
||||
"os"
|
||||
"os/signal"
|
||||
"path/filepath"
|
||||
"syscall"
|
||||
"time"
|
||||
)
|
||||
|
||||
const (
|
||||
every = time.Minute
|
||||
evictEvery = 10 * time.Minute // the cheapest rung still walks directories
|
||||
purgeEvery = 5 * time.Minute
|
||||
)
|
||||
|
||||
type guard struct {
|
||||
log *slog.Logger
|
||||
name string
|
||||
kube *kube
|
||||
ebs *ebs
|
||||
|
||||
evicted, purged time.Time
|
||||
reported float64
|
||||
}
|
||||
|
||||
func main() {
|
||||
log := slog.New(slog.NewJSONHandler(os.Stdout, nil))
|
||||
g := &guard{log: log, name: os.Getenv("NODE_NAME")}
|
||||
if g.name == "" {
|
||||
log.Error("NODE_NAME is unset")
|
||||
os.Exit(1)
|
||||
}
|
||||
k, err := inCluster()
|
||||
if err != nil {
|
||||
log.Error("cluster", "err", err)
|
||||
os.Exit(1)
|
||||
}
|
||||
g.kube = k
|
||||
g.ebs = &ebs{c: &http.Client{Timeout: 30 * time.Second}}
|
||||
|
||||
ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGTERM, syscall.SIGINT)
|
||||
defer stop()
|
||||
if r, err := g.ebs.meta(ctx, "/latest/meta-data/placement/region"); err == nil {
|
||||
g.ebs.region = r
|
||||
} else {
|
||||
log.Warn("no instance metadata: volumes will not grow", "err", err)
|
||||
}
|
||||
t := time.NewTicker(every)
|
||||
defer t.Stop()
|
||||
for {
|
||||
g.tick(ctx)
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case <-t.C:
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (g *guard) read() ([]mount, float64, error) {
|
||||
f, err := os.Open("/proc/1/mountinfo")
|
||||
if err != nil {
|
||||
return nil, 0, err
|
||||
}
|
||||
defer f.Close()
|
||||
ms, err := mounts(f)
|
||||
if err != nil {
|
||||
return nil, 0, err
|
||||
}
|
||||
top := 0.0
|
||||
for i := range ms {
|
||||
if err := measure(&ms[i]); err != nil {
|
||||
return nil, 0, fmt.Errorf("%s: %w", ms[i].Point, err)
|
||||
}
|
||||
top = max(top, ms[i].Fill)
|
||||
}
|
||||
return ms, top, nil
|
||||
}
|
||||
|
||||
func (g *guard) tick(ctx context.Context) {
|
||||
ms, top, err := g.read()
|
||||
if err != nil {
|
||||
g.log.Error("measure", "err", err)
|
||||
return
|
||||
}
|
||||
now := time.Now()
|
||||
if top >= evictAt && now.Sub(g.evicted) >= evictEvery {
|
||||
g.report(ms, "evict")
|
||||
evict(ctx, g.log)
|
||||
g.evicted = now
|
||||
ms, top, _ = g.read()
|
||||
}
|
||||
if top >= purgeAt && now.Sub(g.purged) >= purgeEvery {
|
||||
g.report(ms, "purge")
|
||||
purge(ctx, g.log)
|
||||
g.purged = now
|
||||
ms, top, _ = g.read()
|
||||
}
|
||||
for _, m := range ms {
|
||||
if m.Fill >= purgeAt {
|
||||
g.grow(ctx, m)
|
||||
}
|
||||
g.fit(ctx, m)
|
||||
}
|
||||
if top-g.reported >= 0.01 || g.reported-top >= 0.01 {
|
||||
g.report(ms, "")
|
||||
}
|
||||
g.place(ctx, top)
|
||||
}
|
||||
|
||||
func (g *guard) report(ms []mount, why string) {
|
||||
attrs := []any{"why", why}
|
||||
top := 0.0
|
||||
for _, m := range ms {
|
||||
attrs = append(attrs, m.Point, fmt.Sprintf("%.1f%% of %s", 100*m.Fill, gib(m.Size)))
|
||||
top = max(top, m.Fill)
|
||||
}
|
||||
g.reported = top
|
||||
g.log.Info("disk", attrs...)
|
||||
}
|
||||
|
||||
// place sets the node's label and cordon from the fullest watched filesystem.
|
||||
func (g *guard) place(ctx context.Context, top float64) {
|
||||
n, err := g.kube.node(ctx, g.name)
|
||||
if err != nil {
|
||||
g.log.Error("node", "err", err)
|
||||
return
|
||||
}
|
||||
labels, annotations := map[string]any{}, map[string]any{}
|
||||
patch := map[string]any{}
|
||||
if want := pressured(top, n.Pressure); want != n.Pressure {
|
||||
if want {
|
||||
labels[Label] = Pressure
|
||||
} else {
|
||||
labels[Label] = nil
|
||||
}
|
||||
}
|
||||
switch want := cordoned(top, n.OurCordon); {
|
||||
case want && !n.OurCordon && !n.Cordoned:
|
||||
patch["spec"] = map[string]any{"unschedulable": true}
|
||||
annotations[Cordoned] = "true"
|
||||
case !want && n.OurCordon:
|
||||
if n.Cordoned {
|
||||
patch["spec"] = map[string]any{"unschedulable": false}
|
||||
}
|
||||
annotations[Cordoned] = nil
|
||||
}
|
||||
if len(labels)+len(annotations)+len(patch) == 0 {
|
||||
return
|
||||
}
|
||||
meta := map[string]any{}
|
||||
if len(labels) > 0 {
|
||||
meta["labels"] = labels
|
||||
}
|
||||
if len(annotations) > 0 {
|
||||
meta["annotations"] = annotations
|
||||
}
|
||||
if len(meta) > 0 {
|
||||
patch["metadata"] = meta
|
||||
}
|
||||
if err := g.kube.patch(ctx, g.name, patch); err != nil {
|
||||
g.log.Error("node patch", "err", err)
|
||||
return
|
||||
}
|
||||
g.log.Info("node", "fill", fmt.Sprintf("%.1f%%", 100*top), "labels", labels, "spec", patch["spec"], "annotations", annotations)
|
||||
}
|
||||
|
||||
// grow asks AWS for a larger volume under m, once per cooldown.
|
||||
func (g *guard) grow(ctx context.Context, m mount) {
|
||||
if g.ebs.region == "" {
|
||||
return
|
||||
}
|
||||
b, err := device(host, m.Dev)
|
||||
if err != nil || b.Volume == "" {
|
||||
g.log.Warn("grow: no EBS volume under "+m.Point, "err", err)
|
||||
return
|
||||
}
|
||||
size, err := g.ebs.size(ctx, b.Volume)
|
||||
if err != nil {
|
||||
g.log.Error("grow", "volume", b.Volume, "err", err)
|
||||
return
|
||||
}
|
||||
to, ok := grown(size)
|
||||
if !ok {
|
||||
g.log.Warn("grow: volume at its ceiling", "volume", b.Volume, "gib", size)
|
||||
return
|
||||
}
|
||||
at, state, err := g.ebs.changed(ctx, b.Volume)
|
||||
if err != nil {
|
||||
g.log.Error("grow", "volume", b.Volume, "err", err)
|
||||
return
|
||||
}
|
||||
if since := time.Since(at); since < cooldown || state == "modifying" {
|
||||
g.log.Warn("grow: volume changed recently", "volume", b.Volume, "state", state, "since", since.Round(time.Minute).String())
|
||||
return
|
||||
}
|
||||
if err := g.ebs.grow(ctx, b.Volume, to); err != nil {
|
||||
g.log.Error("grow", "volume", b.Volume, "err", err)
|
||||
return
|
||||
}
|
||||
g.log.Warn("grow", "node", g.name, "mount", m.Point, "volume", b.Volume, "from_gib", size, "to_gib", to, "fill", fmt.Sprintf("%.1f%%", 100*m.Fill))
|
||||
}
|
||||
|
||||
// fit grows the partition and the filesystem under m to the device, once the
|
||||
// device is larger than they are: after a grow, whoever requested it.
|
||||
func (g *guard) fit(ctx context.Context, m mount) {
|
||||
b, err := device(host, m.Dev)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
const slack = 128 << 20
|
||||
if b.Part > 0 && b.Room > slack {
|
||||
out, err := run(ctx, "/usr/bin/growpart", "/dev/"+b.Disk, fmt.Sprint(b.Part))
|
||||
g.log.Warn("growpart", "dev", b.Name, "err", err, "out", last(out))
|
||||
if b, err = device(host, m.Dev); err != nil {
|
||||
return
|
||||
}
|
||||
}
|
||||
size, err := fsBytes(filepath.Join(host, "dev", b.Name), m.Type)
|
||||
if err != nil || size+slack >= b.Bytes {
|
||||
return
|
||||
}
|
||||
var argv []string
|
||||
switch m.Type {
|
||||
case "xfs":
|
||||
argv = []string{"/usr/sbin/xfs_growfs", m.Point}
|
||||
default:
|
||||
argv = []string{"/usr/sbin/resize2fs", "/dev/" + b.Name}
|
||||
}
|
||||
out, err := run(ctx, argv...)
|
||||
g.log.Warn("resize", "mount", m.Point, "dev", b.Name, "from", gib(size), "to", gib(b.Bytes), "err", err, "out", last(out))
|
||||
}
|
||||
|
||||
func gib(b uint64) string { return fmt.Sprintf("%.0fG", float64(b)/(1<<30)) }
|
||||
+113
@@ -0,0 +1,113 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"bytes"
|
||||
"context"
|
||||
"crypto/tls"
|
||||
"crypto/x509"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"net"
|
||||
"net/http"
|
||||
"os"
|
||||
"strings"
|
||||
"time"
|
||||
)
|
||||
|
||||
// kube is the cluster API as this pod's service account, which may get and patch
|
||||
// nodes and nothing else (infra/aws/k8s/disk.yaml).
|
||||
type kube struct {
|
||||
base string
|
||||
c *http.Client
|
||||
}
|
||||
|
||||
const account = "/var/run/secrets/kubernetes.io/serviceaccount/"
|
||||
|
||||
func inCluster() (*kube, error) {
|
||||
h, p := os.Getenv("KUBERNETES_SERVICE_HOST"), os.Getenv("KUBERNETES_SERVICE_PORT")
|
||||
if h == "" || p == "" {
|
||||
return nil, errors.New("not in a cluster: KUBERNETES_SERVICE_HOST is unset")
|
||||
}
|
||||
ca, err := os.ReadFile(account + "ca.crt")
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
pool := x509.NewCertPool()
|
||||
if !pool.AppendCertsFromPEM(ca) {
|
||||
return nil, errors.New("service account ca.crt holds no certificate")
|
||||
}
|
||||
return &kube{
|
||||
base: "https://" + net.JoinHostPort(h, p),
|
||||
c: &http.Client{Timeout: 20 * time.Second, Transport: &http.Transport{
|
||||
TLSClientConfig: &tls.Config{RootCAs: pool, MinVersion: tls.VersionTLS12},
|
||||
}},
|
||||
}, nil
|
||||
}
|
||||
|
||||
// nodeState is what this program reads of its node.
|
||||
type nodeState struct {
|
||||
Pressure bool // carries Label=Pressure
|
||||
Cordoned bool // spec.unschedulable
|
||||
OurCordon bool // the cordon is this program's (annotation Cordoned)
|
||||
}
|
||||
|
||||
func (k *kube) node(ctx context.Context, name string) (nodeState, error) {
|
||||
var n struct {
|
||||
Metadata struct {
|
||||
Labels map[string]string `json:"labels"`
|
||||
Annotations map[string]string `json:"annotations"`
|
||||
} `json:"metadata"`
|
||||
Spec struct {
|
||||
Unschedulable bool `json:"unschedulable"`
|
||||
} `json:"spec"`
|
||||
}
|
||||
if err := k.do(ctx, http.MethodGet, "/api/v1/nodes/"+name, nil, &n); err != nil {
|
||||
return nodeState{}, err
|
||||
}
|
||||
return nodeState{
|
||||
Pressure: n.Metadata.Labels[Label] == Pressure,
|
||||
Cordoned: n.Spec.Unschedulable,
|
||||
OurCordon: n.Metadata.Annotations[Cordoned] == "true",
|
||||
}, nil
|
||||
}
|
||||
|
||||
// patch applies a JSON merge patch to the node; a null value deletes a key.
|
||||
func (k *kube) patch(ctx context.Context, name string, body map[string]any) error {
|
||||
b, err := json.Marshal(body)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
return k.do(ctx, http.MethodPatch, "/api/v1/nodes/"+name, b, nil)
|
||||
}
|
||||
|
||||
func (k *kube) do(ctx context.Context, method, path string, body []byte, out any) error {
|
||||
// Re-read every call: the projected token is rotated under a running pod.
|
||||
tok, err := os.ReadFile(account + "token")
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
req, err := http.NewRequestWithContext(ctx, method, k.base+path, bytes.NewReader(body))
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
req.Header.Set("Authorization", "Bearer "+strings.TrimSpace(string(tok)))
|
||||
req.Header.Set("Accept", "application/json")
|
||||
if body != nil {
|
||||
req.Header.Set("Content-Type", "application/merge-patch+json")
|
||||
}
|
||||
resp, err := k.c.Do(req)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
data, _ := io.ReadAll(io.LimitReader(resp.Body, 1<<20))
|
||||
if resp.StatusCode/100 != 2 {
|
||||
return fmt.Errorf("%s %s: %s: %s", method, path, resp.Status, strings.TrimSpace(string(data)))
|
||||
}
|
||||
if out != nil {
|
||||
return json.Unmarshal(data, out)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
@@ -0,0 +1,55 @@
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"os/exec"
|
||||
"runtime"
|
||||
"strings"
|
||||
"syscall"
|
||||
)
|
||||
|
||||
// run executes a program of the NODE's, in the node's mount namespace: its
|
||||
// crictl, journalctl, growpart and resize2fs, against its own /dev and /run. The
|
||||
// image carries none of them.
|
||||
//
|
||||
// setns(CLONE_NEWNS) is refused to a thread that shares its filesystem
|
||||
// attributes, and every Go thread shares them, so the call runs on a thread of
|
||||
// its own that first unshares them. That thread is never unlocked: it leaves the
|
||||
// node's namespace only by exiting, which the runtime does when a goroutine ends
|
||||
// still locked. The child inherits the namespace from the thread that forks it.
|
||||
func run(ctx context.Context, argv ...string) (string, error) {
|
||||
type result struct {
|
||||
out string
|
||||
err error
|
||||
}
|
||||
done := make(chan result, 1)
|
||||
go func() {
|
||||
runtime.LockOSThread()
|
||||
if err := syscall.Unshare(syscall.CLONE_FS); err != nil {
|
||||
done <- result{err: fmt.Errorf("unshare: %w", err)}
|
||||
return
|
||||
}
|
||||
fd, err := syscall.Open("/proc/1/ns/mnt", syscall.O_RDONLY|syscall.O_CLOEXEC, 0)
|
||||
if err != nil {
|
||||
done <- result{err: err}
|
||||
return
|
||||
}
|
||||
_, _, e := syscall.RawSyscall(sysSetns, uintptr(fd), syscall.CLONE_NEWNS, 0)
|
||||
syscall.Close(fd)
|
||||
if e != 0 {
|
||||
done <- result{err: fmt.Errorf("setns: %w", e)}
|
||||
return
|
||||
}
|
||||
cmd := exec.CommandContext(ctx, argv[0], argv[1:]...)
|
||||
cmd.Env = []string{"PATH=/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin", "LANG=C"}
|
||||
cmd.Dir = "/"
|
||||
out, err := cmd.CombinedOutput()
|
||||
if len(out) > 4096 {
|
||||
out = out[len(out)-4096:]
|
||||
}
|
||||
done <- result{strings.TrimSpace(string(out)), err}
|
||||
}()
|
||||
r := <-done
|
||||
return r.out, r.err
|
||||
}
|
||||
@@ -0,0 +1,13 @@
|
||||
//go:build !linux
|
||||
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
)
|
||||
|
||||
// run needs Linux namespaces; the program serves only Linux nodes.
|
||||
func run(ctx context.Context, argv ...string) (string, error) {
|
||||
return "", errors.New("disk runs node programs only on linux")
|
||||
}
|
||||
@@ -0,0 +1,6 @@
|
||||
//go:build linux
|
||||
|
||||
package main
|
||||
|
||||
// sysSetns is setns(2), which the frozen syscall package does not number.
|
||||
const sysSetns = 308
|
||||
@@ -0,0 +1,6 @@
|
||||
//go:build linux
|
||||
|
||||
package main
|
||||
|
||||
// sysSetns is setns(2), which the frozen syscall package does not number.
|
||||
const sysSetns = 268
|
||||
@@ -1,15 +1,28 @@
|
||||
# Canonical CI/CD config for hanzoai/k8s — read by the hanzoai/ci reusable workflow
|
||||
# (.hanzo/workflows/cicd.yml) and by platform.hanzo.ai.
|
||||
#
|
||||
# This is a zip PLUGIN, so it ships as a BINARY a running host installs and verifies by
|
||||
# digest, not as an image a cluster runs. There is no images: lane and no deploy: lane:
|
||||
# the composition that serves /v1/k8s is hanzoai/cloud's, and it installs this artifact
|
||||
# by URL + SHA-256 from the index published below.
|
||||
# The k8s service is a zip PLUGIN, so it ships as a BINARY a running host installs and
|
||||
# verifies by digest, not as an image a cluster runs: the composition that serves
|
||||
# /v1/k8s is hanzoai/cloud's, and it installs this artifact by URL + SHA-256 from the
|
||||
# index published below.
|
||||
#
|
||||
# disk/ is the one image: the node agent that keeps every node's disks from filling,
|
||||
# run as a DaemonSet (universe infra/aws/k8s/disk.yaml). Its own module, standard
|
||||
# library only, so the plugin's gates and graph never see it.
|
||||
binaries:
|
||||
- name: k8s
|
||||
main: .
|
||||
platforms: [linux/amd64, linux/arm64]
|
||||
|
||||
images:
|
||||
- name: disk
|
||||
context: disk
|
||||
dockerfile: disk/Dockerfile
|
||||
repo: ghcr.io/hanzoai/disk
|
||||
platforms:
|
||||
- linux/amd64
|
||||
- linux/arm64
|
||||
|
||||
test:
|
||||
- name: go-test
|
||||
# `make test`, never a bare `go test`: the target carries `go vet` and the four
|
||||
@@ -19,3 +32,9 @@ test:
|
||||
set -e
|
||||
export GOPRIVATE='github.com/hanzoai/*' GOWORK=off
|
||||
make test
|
||||
- name: disk
|
||||
run: |
|
||||
set -e
|
||||
cd disk
|
||||
GOWORK=off go vet ./...
|
||||
GOWORK=off go test ./... -count=1
|
||||
|
||||
Reference in New Issue
Block a user