Commit fb2b232
Eric Bower
·
2026-08-09 11:36:50 -0400 EDT
parent 2064538
fix(gc): wasn't killing zmx sessions properly
2 files changed,
+70,
-28
M
main.go
M
main.go
+42,
-20
| ... | ... | @@ -666,9 +666,6 @@ func getDomain(eventType string) string { | |
| 666 | 666 | if eventType == "local" { | |
| 667 | 667 | return "local" | |
| 668 | 668 | } | |
| 669 | - | if eventType == "test-ci" || strings.HasPrefix(eventType, "test-") { | |
| 670 | - | return eventType | |
| 671 | - | } | |
| 672 | 669 | return "ci" | |
| 673 | 670 | } | |
| 674 | 671 |
| ... | ... | @@ -1919,28 +1916,29 @@ func cancelJobSessions(prefix string) { | |
| 1919 | 1916 | } | |
| 1920 | 1917 | } | |
| 1921 | 1918 | ||
| 1922 | - | // runGC kills all ci. zmx sessions older than 3 hours, regardless of status. | |
| 1923 | - | func runGC(cfg *Cfg) error { | |
| 1924 | - | log := cfg.Logger.With("cmd", "gc") | |
| 1925 | - | log.Debug("running garbage collection") | |
| 1926 | - | ||
| 1927 | - | listOutput, err := exec.Command("zmx", "list").CombinedOutput() | |
| 1928 | - | if err != nil { | |
| 1929 | - | return fmt.Errorf("zmx list: %w", err) | |
| 1930 | - | } | |
| 1931 | - | ||
| 1932 | - | sessions := parseZMXList(string(listOutput)) | |
| 1933 | - | ||
| 1934 | - | cutoff := time.Now().Add(-3 * time.Hour).Unix() | |
| 1935 | - | ||
| 1919 | + | // findSessionsForGC identifies finished sessions or sessions older than 3 hours matching given prefixes. | |
| 1920 | + | func findSessionsForGC(sessions []SessionInfo, prefixes []string, now time.Time) []string { | |
| 1921 | + | cutoff := now.Add(-3 * time.Hour).Unix() | |
| 1936 | 1922 | var toKill []string | |
| 1937 | 1923 | for _, s := range sessions { | |
| 1938 | - | if !strings.HasPrefix(s.Name, "ci.") && !strings.HasPrefix(s.Name, "local.") { | |
| 1924 | + | matched := false | |
| 1925 | + | for _, prefix := range prefixes { | |
| 1926 | + | if strings.HasPrefix(s.Name, prefix) { | |
| 1927 | + | matched = true | |
| 1928 | + | break | |
| 1929 | + | } | |
| 1930 | + | } | |
| 1931 | + | if !matched { | |
| 1932 | + | continue | |
| 1933 | + | } | |
| 1934 | + | ||
| 1935 | + | if s.Ended != "" { | |
| 1936 | + | toKill = append(toKill, s.Name) | |
| 1939 | 1937 | continue | |
| 1940 | 1938 | } | |
| 1941 | 1939 | ||
| 1942 | 1940 | if s.Created == "" { | |
| 1943 | - | continue // skip sessions with no creation time | |
| 1941 | + | continue | |
| 1944 | 1942 | } | |
| 1945 | 1943 | ||
| 1946 | 1944 | var created int64 |
| ... | ... | @@ -1949,10 +1947,34 @@ func runGC(cfg *Cfg) error { | |
| 1949 | 1947 | } | |
| 1950 | 1948 | ||
| 1951 | 1949 | if created < cutoff { | |
| 1952 | - | log.Debug("session expired, scheduling for gc", "session", s.Name, "created", s.Created) | |
| 1953 | 1950 | toKill = append(toKill, s.Name) | |
| 1954 | 1951 | } | |
| 1955 | 1952 | } | |
| 1953 | + | return toKill | |
| 1954 | + | } | |
| 1955 | + | ||
| 1956 | + | // runGC kills finished zmx sessions or sessions older than 3 hours. | |
| 1957 | + | func runGC(cfg *Cfg) error { | |
| 1958 | + | log := cfg.Logger.With("cmd", "gc") | |
| 1959 | + | log.Debug("running garbage collection") | |
| 1960 | + | ||
| 1961 | + | listOutput, err := exec.Command("zmx", "list").CombinedOutput() | |
| 1962 | + | if err != nil { | |
| 1963 | + | return fmt.Errorf("zmx list: %w", err) | |
| 1964 | + | } | |
| 1965 | + | ||
| 1966 | + | sessions := parseZMXList(string(listOutput)) | |
| 1967 | + | ||
| 1968 | + | prefixes := []string{"ci.", "local."} | |
| 1969 | + | if cfg != nil && cfg.SessionPrefix != "" { | |
| 1970 | + | p := cfg.SessionPrefix | |
| 1971 | + | if !strings.HasSuffix(p, ".") { | |
| 1972 | + | p += "." | |
| 1973 | + | } | |
| 1974 | + | prefixes = append(prefixes, p) | |
| 1975 | + | } | |
| 1976 | + | ||
| 1977 | + | toKill := findSessionsForGC(sessions, prefixes, time.Now()) | |
| 1956 | 1978 | ||
| 1957 | 1979 | if len(toKill) == 0 { | |
| 1958 | 1980 | log.Debug("no sessions to garbage collect") |
+28,
-8
| ... | ... | @@ -48,10 +48,10 @@ zmx run step2 echo "hello from step2" | |
| 48 | 48 | ctx, cancel := context.WithCancel(context.Background()) | |
| 49 | 49 | testJobID := fmt.Sprintf("j-%d", time.Now().UnixNano()%100000000) | |
| 50 | 50 | t.Cleanup(func() { | |
| 51 | - | _ = exec.Command("zmx", "kill", "-f", fmt.Sprintf("test-ci.test-repo.%s", testJobID)).Run() | |
| 51 | + | _ = exec.Command("zmx", "kill", "-f", fmt.Sprintf("local.test-repo.%s", testJobID)).Run() | |
| 52 | 52 | }) | |
| 53 | 53 | event := Event{ | |
| 54 | - | Type: "test-ci", | |
| 54 | + | Type: "local", | |
| 55 | 55 | Name: "test-repo", | |
| 56 | 56 | JobID: testJobID, | |
| 57 | 57 | Workspace: workspaceDir, |
| ... | ... | @@ -68,7 +68,7 @@ zmx run step2 echo "hello from step2" | |
| 68 | 68 | NewWorkspace: defaultWorkspaceFactory, | |
| 69 | 69 | StatusOutput: statusBuf, | |
| 70 | 70 | IncludeRunning: true, | |
| 71 | - | SessionPrefix: "test-ci.", | |
| 71 | + | SessionPrefix: "local.", | |
| 72 | 72 | } | |
| 73 | 73 | ||
| 74 | 74 | // 3. Run the runner (fire-and-forget, exits quickly) |
| ... | ... | @@ -185,8 +185,8 @@ zmx run step2 echo "hello from step2" | |
| 185 | 185 | if finalPayload == nil { | |
| 186 | 186 | t.Fatal("no final payload") | |
| 187 | 187 | } | |
| 188 | - | // Filter to only sessions with our job prefix to avoid picking up unrelated test-ci.* sessions | |
| 189 | - | expectedPrefix := fmt.Sprintf("test-ci.%s.%s.", finalPayload.Name, finalPayload.JobID) | |
| 188 | + | // Filter to only sessions with our job prefix to avoid picking up unrelated local.* sessions | |
| 189 | + | expectedPrefix := fmt.Sprintf("local.%s.%s.", finalPayload.Name, finalPayload.JobID) | |
| 190 | 190 | sessions := make([]SessionInfo, 0, len(finalPayload.Sessions)) | |
| 191 | 191 | for _, s := range finalPayload.Sessions { | |
| 192 | 192 | if strings.HasPrefix(s.Name, expectedPrefix) { |
| ... | ... | @@ -559,7 +559,6 @@ func TestGetDomain(t *testing.T) { | |
| 559 | 559 | {"ci", "ci"}, | |
| 560 | 560 | {"", "ci"}, | |
| 561 | 561 | {"local", "local"}, | |
| 562 | - | {"test-ci", "test-ci"}, | |
| 563 | 562 | } | |
| 564 | 563 | ||
| 565 | 564 | for _, tt := range tests { |
| ... | ... | @@ -639,6 +638,27 @@ func findRunningJobsFromOutput(output, name string) ([]string, []SessionInfo) { | |
| 639 | 638 | return runners, sessions | |
| 640 | 639 | } | |
| 641 | 640 | ||
| 641 | + | func TestFindSessionsForGC(t *testing.T) { | |
| 642 | + | now := time.Unix(1786289700, 0) | |
| 643 | + | twoHoursAgo := fmt.Sprintf("%d", now.Add(-2*time.Hour).Unix()) | |
| 644 | + | fourHoursAgo := fmt.Sprintf("%d", now.Add(-4*time.Hour).Unix()) | |
| 645 | + | ||
| 646 | + | sessions := []SessionInfo{ | |
| 647 | + | {Name: "ci.repo.active.runner", Created: twoHoursAgo, Ended: ""}, | |
| 648 | + | {Name: "ci.repo.finished.runner", Created: twoHoursAgo, Ended: "1786289600"}, | |
| 649 | + | {Name: "ci.repo.expired.runner", Created: fourHoursAgo, Ended: ""}, | |
| 650 | + | {Name: "other.session", Created: fourHoursAgo, Ended: ""}, | |
| 651 | + | } | |
| 652 | + | ||
| 653 | + | toKill := findSessionsForGC(sessions, []string{"ci.", "local."}, now) | |
| 654 | + | if len(toKill) != 2 { | |
| 655 | + | t.Fatalf("expected 2 sessions for GC, got %d: %v", len(toKill), toKill) | |
| 656 | + | } | |
| 657 | + | if toKill[0] != "ci.repo.finished.runner" || toKill[1] != "ci.repo.expired.runner" { | |
| 658 | + | t.Errorf("unexpected sessions for GC: %v", toKill) | |
| 659 | + | } | |
| 660 | + | } | |
| 661 | + | ||
| 642 | 662 | func TestKillSessionsEmpty(t *testing.T) { | |
| 643 | 663 | // Should not error with empty list | |
| 644 | 664 | if err := killSessions(nil); err != nil { |
| ... | ... | @@ -795,7 +815,7 @@ func TestWaitAndReport_Cancellation(t *testing.T) { | |
| 795 | 815 | } | |
| 796 | 816 | ||
| 797 | 817 | jobID := fmt.Sprintf("canceltest-%d", time.Now().UnixNano()) | |
| 798 | - | prefix := "test-local.testrepo." + jobID + "." | |
| 818 | + | prefix := "local.testrepo." + jobID + "." | |
| 799 | 819 | runnerSession := prefix + "runner" | |
| 800 | 820 | ||
| 801 | 821 | t.Cleanup(func() { |
| ... | ... | @@ -812,7 +832,7 @@ func TestWaitAndReport_Cancellation(t *testing.T) { | |
| 812 | 832 | cancel() | |
| 813 | 833 | }() | |
| 814 | 834 | ||
| 815 | - | err := waitAndReport(cfg, nil, "testrepo", jobID, "test-local") | |
| 835 | + | err := waitAndReport(cfg, nil, "testrepo", jobID, "local") | |
| 816 | 836 | if err == nil { | |
| 817 | 837 | t.Error("expected error when waitAndReport is cancelled, got nil") | |
| 818 | 838 | } |