internal/logs/many_test.go
Rama main · réplica de lectura del repositorio original.
package logs
import (
"context"
"encoding/json"
"fmt"
"net/http"
"slices"
"strings"
"sync"
"testing"
"time"
corev1 "k8s.io/api/core/v1"
"k8s.io/apimachinery/pkg/types"
"k8s.io/client-go/kubernetes"
)
type cluster struct {
mu sync.Mutex
pods []string
lines map[string][]string
listed int
selector string
hold bool
tail string
}
func (apiserver *cluster) handler(t *testing.T) http.HandlerFunc {
t.Helper()
return func(w http.ResponseWriter, r *http.Request) {
if strings.HasSuffix(r.URL.Path, "/log") {
apiserver.serveLog(w, r)
return
}
apiserver.serveList(w, r)
}
}
func (apiserver *cluster) serveList(w http.ResponseWriter, r *http.Request) {
apiserver.mu.Lock()
apiserver.listed++
apiserver.selector = r.URL.Query().Get("labelSelector")
names := append([]string{}, apiserver.pods...)
apiserver.mu.Unlock()
list := corev1.PodList{}
for _, name := range names {
list.Items = append(list.Items, corev1.Pod{
Name: name, Namespace: "prod",
})
}
w.Header().Set("Content-Type", "application/json")
_ = json.NewEncoder(w).Encode(list)
}
func (apiserver *cluster) serveLog(w http.ResponseWriter, r *http.Request) {
parts := strings.Split(strings.TrimPrefix(r.URL.Path, "/"), "/")
name := parts[len(parts)-2]
apiserver.mu.Lock()
lines := apiserver.lines[name]
apiserver.tail = r.URL.Query().Get("tailLines")
apiserver.mu.Unlock()
for _, line := range lines {
_, _ = fmt.Fprintln(w, line)
}
if !apiserver.holding() {
return
}
flusher, ok := w.(http.Flusher)
if ok {
flusher.Flush()
}
<-r.Context().Done()
}
func (apiserver *cluster) holdOpen() {
apiserver.mu.Lock()
defer apiserver.mu.Unlock()
apiserver.hold = true
}
func (apiserver *cluster) holding() bool {
apiserver.mu.Lock()
defer apiserver.mu.Unlock()
return apiserver.hold
}
func (apiserver *cluster) add(name string, lines ...string) {
apiserver.mu.Lock()
defer apiserver.mu.Unlock()
apiserver.pods = append(apiserver.pods, name)
apiserver.lines[name] = lines
}
func (apiserver *cluster) askedForTail() string {
apiserver.mu.Lock()
defer apiserver.mu.Unlock()
return apiserver.tail
}
func (apiserver *cluster) lists() int {
apiserver.mu.Lock()
defer apiserver.mu.Unlock()
return apiserver.listed
}
func (apiserver *cluster) askedFor() string {
apiserver.mu.Lock()
defer apiserver.mu.Unlock()
return apiserver.selector
}
func newCluster() *cluster {
return &cluster{lines: map[string][]string{}}
}
func manyClient(t *testing.T, apiserver *cluster) kubernetes.Interface {
t.Helper()
return clientFor(t, apiserver.handler(t))
}
func gather(t *testing.T, stream *Stream, want int) map[string][]string {
t.Helper()
out := map[string][]string{}
count := 0
deadline := time.After(5 * time.Second)
for count < want {
select {
case line, ok := <-stream.Lines:
if !ok {
return out
}
out[line.Pod] = append(out[line.Pod], line.Text)
count++
case <-deadline:
t.Fatalf("timed out after %d lines: %v", count, out)
}
}
return out
}
func TestEveryMatchingPodIsRead(t *testing.T) {
apiserver := newCluster()
apiserver.add("web-0", "from web-0")
apiserver.add("web-1", "from web-1")
apiserver.add("web-2", "from web-2")
stream, err := Open(t.Context(), manyClient(t, apiserver), Request{
Namespace: "prod",
Selector: "app=web",
})
if err != nil {
t.Fatalf("open: %v", err)
}
defer stream.Close()
found := gather(t, stream, 3)
if len(found) != 3 {
t.Fatalf("pods = %v, want all three read", found)
}
for _, name := range []string{"web-0", "web-1", "web-2"} {
if found[name][0] != "from "+name {
t.Fatalf("%s said %v", name, found[name])
}
}
}
func TestALineSaysWhichPodWroteIt(t *testing.T) {
apiserver := newCluster()
apiserver.add("web-0", "hello")
stream, err := Open(t.Context(), manyClient(t, apiserver), Request{Namespace: "prod", Selector: "app=web"})
if err != nil {
t.Fatalf("open: %v", err)
}
defer stream.Close()
found := gather(t, stream, 1)
if found["web-0"] == nil {
t.Fatalf("lines = %v, want them tagged with the pod", found)
}
}
func TestTheSelectorIsWhatTheApiserverIsAsked(t *testing.T) {
apiserver := newCluster()
apiserver.add("web-0", "hello")
stream, err := Open(t.Context(), manyClient(t, apiserver), Request{
Namespace: "prod",
Selector: "app=web,tier=front",
})
if err != nil {
t.Fatalf("open: %v", err)
}
defer stream.Close()
gather(t, stream, 1)
if apiserver.askedFor() != "app=web,tier=front" {
t.Fatalf("selector = %q", apiserver.askedFor())
}
}
func TestASelectorThatMatchesNothingSaysSo(t *testing.T) {
apiserver := newCluster()
_, err := Open(t.Context(), manyClient(t, apiserver), Request{Namespace: "prod", Selector: "app=gone"})
if err == nil {
t.Fatal("a selector matching no pods opened a stream that would never say anything")
}
if !strings.Contains(err.Error(), "no pods match") {
t.Fatalf("error = %v", err)
}
}
func TestOnlySoManyPodsAreOpenedAtOnce(t *testing.T) {
apiserver := newCluster()
for i := range maxPods + 5 {
apiserver.add(fmt.Sprintf("web-%02d", i), "hello")
}
stream, err := Open(t.Context(), manyClient(t, apiserver), Request{Namespace: "prod", Selector: "app=web"})
if err != nil {
t.Fatalf("open: %v", err)
}
defer stream.Close()
if stream.Attached() != maxPods {
t.Fatalf("attached = %d, want the cap", stream.Attached())
}
if stream.Matched() != maxPods+5 {
t.Fatalf("matched = %d, want every pod counted", stream.Matched())
}
}
func TestAMergedStreamEndsWhenEveryPodIsDone(t *testing.T) {
apiserver := newCluster()
apiserver.add("web-0", "one")
apiserver.add("web-1", "two")
stream, err := Open(t.Context(), manyClient(t, apiserver), Request{Namespace: "prod", Selector: "app=web"})
if err != nil {
t.Fatalf("open: %v", err)
}
defer stream.Close()
seen := 0
for range stream.Lines {
seen++
}
if seen != 2 {
t.Fatalf("lines = %d, want both before the channel closed", seen)
}
}
func TestAPodThatAppearsLaterIsPickedUp(t *testing.T) {
restore := hurryResolve(t)
defer restore()
apiserver := newCluster()
apiserver.add("web-0", "from web-0")
stream, err := Open(t.Context(), manyClient(t, apiserver), Request{
Namespace: "prod",
Selector: "app=web",
Follow: true,
})
if err != nil {
t.Fatalf("open: %v", err)
}
defer stream.Close()
gather(t, stream, 1)
apiserver.add("web-1", "from web-1")
found := gather(t, stream, 1)
if found["web-1"] == nil {
t.Fatalf("lines = %v, want the pod that appeared after the stream opened", found)
}
}
func TestAPodThatAppearsLaterAndRefusesLogsIsReported(t *testing.T) {
restore := hurryResolve(t)
defer restore()
apiserver := newCluster()
apiserver.holdOpen()
apiserver.add("web-0", "from web-0")
client := clientFor(t, func(w http.ResponseWriter, r *http.Request) {
if strings.Contains(r.URL.Path, "web-1") && strings.HasSuffix(r.URL.Path, "/log") {
w.WriteHeader(http.StatusForbidden)
_, _ = w.Write([]byte(`{"kind":"Status","message":"pods/log is forbidden","code":403}`))
return
}
apiserver.handler(t)(w, r)
})
stream, err := Open(t.Context(), client, Request{
Namespace: "prod",
Selector: "app=web",
Follow: true,
})
if err != nil {
t.Fatalf("open: %v", err)
}
defer stream.Close()
gather(t, stream, 1)
apiserver.add("web-1", "from web-1")
deadline := time.Now().Add(3 * time.Second)
for time.Now().Before(deadline) {
if stream.Matched() == 2 && stream.Err() != nil {
break
}
time.Sleep(10 * time.Millisecond)
}
if stream.Attached() != 1 || stream.Matched() != 2 {
t.Fatalf("attached %d of %d, want the original pod only", stream.Attached(), stream.Matched())
}
if stream.Err() == nil || !strings.Contains(stream.Err().Error(), "web-1") {
t.Fatalf("error = %v, want the refusing pod named", stream.Err())
}
}
func TestTheCountOfPodsFollowsTheRollout(t *testing.T) {
restore := hurryResolve(t)
defer restore()
apiserver := newCluster()
apiserver.holdOpen()
apiserver.add("web-0", "from web-0")
stream, err := Open(t.Context(), manyClient(t, apiserver), Request{
Namespace: "prod",
Selector: "app=web",
Follow: true,
})
if err != nil {
t.Fatalf("open: %v", err)
}
defer stream.Close()
gather(t, stream, 1)
if stream.Attached() != 1 || stream.Matched() != 1 {
t.Fatalf("attached %d of %d, want 1 of 1 at open", stream.Attached(), stream.Matched())
}
apiserver.add("web-1", "from web-1")
gather(t, stream, 1)
deadline := time.Now().Add(3 * time.Second)
for time.Now().Before(deadline) {
if stream.Attached() == 2 && stream.Matched() == 2 {
return
}
time.Sleep(10 * time.Millisecond)
}
t.Fatalf(
"attached %d of %d, want the second pod counted once it is being read",
stream.Attached(),
stream.Matched(),
)
}
func TestAFollowingStreamKeepsAskingWhoElseIsThere(t *testing.T) {
restore := hurryResolve(t)
defer restore()
apiserver := newCluster()
apiserver.add("web-0", "hello")
stream, err := Open(t.Context(), manyClient(t, apiserver), Request{
Namespace: "prod",
Selector: "app=web",
Follow: true,
})
if err != nil {
t.Fatalf("open: %v", err)
}
defer stream.Close()
gather(t, stream, 1)
deadline := time.Now().Add(3 * time.Second)
for time.Now().Before(deadline) {
if apiserver.lists() > 1 {
return
}
time.Sleep(10 * time.Millisecond)
}
t.Fatalf("the pod list was read %d times, so a rollout would go unnoticed", apiserver.lists())
}
func TestClosingAMergedStreamStopsEveryPod(t *testing.T) {
apiserver := newCluster()
apiserver.add("web-0", "hello")
apiserver.add("web-1", "hello")
stream, err := Open(context.Background(), manyClient(t, apiserver), Request{
Namespace: "prod",
Selector: "app=web",
Follow: true,
})
if err != nil {
t.Fatalf("open: %v", err)
}
gather(t, stream, 2)
stream.Close()
deadline := time.After(5 * time.Second)
for {
select {
case _, ok := <-stream.Lines:
if !ok {
return
}
case <-deadline:
t.Fatal("the merged channel never closed after the stream was closed")
}
}
}
func TestClosingAMergedStreamUnblocksAFullOutputBuffer(t *testing.T) {
apiserver := newCluster()
apiserver.add("web-0", strings.Repeat("line\n", lineBuffer*4))
stream, err := Open(t.Context(), manyClient(t, apiserver), Request{
Namespace: "prod",
Selector: "app=web",
})
if err != nil {
t.Fatalf("open: %v", err)
}
deadline := time.Now().Add(3 * time.Second)
for len(stream.Lines) < lineBuffer*2 && time.Now().Before(deadline) {
time.Sleep(time.Millisecond)
}
if len(stream.Lines) != lineBuffer*2 {
t.Fatalf("buffered lines = %d, want the merged buffer full", len(stream.Lines))
}
stream.Close()
drained := make(chan int, 1)
go func() {
count := 0
for range stream.Lines {
count++
}
drained <- count
}()
select {
case count := <-drained:
if count < lineBuffer*2 {
t.Fatalf("drained lines = %d, want at least the full output buffer", count)
}
case <-time.After(3 * time.Second):
t.Fatal("the merged stream stayed blocked after it was closed")
}
}
func hurryResolve(t *testing.T) func() {
t.Helper()
old := resolveEvery
resolveEvery = 20 * time.Millisecond
return func() {
resolveEvery = old
}
}
func TestAPodThatIsNotReadyYetIsAskedAgain(t *testing.T) {
restore := hurryResolve(t)
defer restore()
apiserver := newCluster()
apiserver.add("web-0", "ready now")
refusals := 0
var mu sync.Mutex
client := clientFor(t, func(w http.ResponseWriter, r *http.Request) {
if strings.HasSuffix(r.URL.Path, "/log") {
mu.Lock()
first := refusals == 0
refusals++
mu.Unlock()
if first {
w.WriteHeader(http.StatusBadRequest)
return
}
}
apiserver.handler(t)(w, r)
})
stream, err := Open(t.Context(), client, Request{
Namespace: "prod",
Selector: "app=web",
Follow: true,
})
if err != nil {
t.Fatalf("open: %v", err)
}
defer stream.Close()
found := gather(t, stream, 1)
if found["web-0"] == nil {
t.Fatalf("lines = %v, want the pod read once it was ready", found)
}
}
func TestAFollowingStreamReportsASelectorRefreshFailure(t *testing.T) {
restore := hurryResolve(t)
defer restore()
apiserver := newCluster()
apiserver.holdOpen()
apiserver.add("web-0", "ready")
var mu sync.Mutex
listed := 0
client := clientFor(t, func(w http.ResponseWriter, r *http.Request) {
if !strings.HasSuffix(r.URL.Path, "/log") {
mu.Lock()
listed++
current := listed
mu.Unlock()
if current > 1 {
w.WriteHeader(http.StatusForbidden)
_, _ = w.Write([]byte(`{"kind":"Status","message":"pods are forbidden","code":403}`))
return
}
}
apiserver.handler(t)(w, r)
})
stream, err := Open(t.Context(), client, Request{
Namespace: "prod",
Selector: "app=web",
Follow: true,
})
if err != nil {
t.Fatalf("open: %v", err)
}
defer stream.Close()
deadline := time.Now().Add(time.Second)
for stream.Err() == nil && time.Now().Before(deadline) {
time.Sleep(time.Millisecond)
}
if stream.Err() == nil || !strings.Contains(stream.Err().Error(), "refresh pods matching") {
t.Fatalf("error = %v, want the failed selector refresh reported", stream.Err())
}
}
func TestAPodWithSeveralContainersIsAskedAboutOneOfThem(t *testing.T) {
asked := make(chan string, 4)
client := clientFor(t, func(w http.ResponseWriter, r *http.Request) {
if strings.HasSuffix(r.URL.Path, "/log") {
asked <- r.URL.Query().Get("container")
_, _ = fmt.Fprintln(w, "hello")
return
}
list := corev1.PodList{Items: []corev1.Pod{{
Name: "web-0", Namespace: "prod",
Spec: corev1.PodSpec{Containers: []corev1.Container{
{Name: "app"},
{Name: "sidecar"},
}},
}}}
w.Header().Set("Content-Type", "application/json")
_ = json.NewEncoder(w).Encode(list)
})
stream, err := Open(t.Context(), client, Request{Namespace: "prod", Selector: "app=web"})
if err != nil {
t.Fatalf("open: %v", err)
}
defer stream.Close()
gather(t, stream, 1)
select {
case container := <-asked:
if container != "app" {
t.Fatalf("container = %q, want the pod's first, since naming none is refused", container)
}
case <-time.After(5 * time.Second):
t.Fatal("no log request was made")
}
}
func TestThePodsOwnDefaultContainerWins(t *testing.T) {
asked := make(chan string, 4)
client := clientFor(t, func(w http.ResponseWriter, r *http.Request) {
if strings.HasSuffix(r.URL.Path, "/log") {
asked <- r.URL.Query().Get("container")
_, _ = fmt.Fprintln(w, "hello")
return
}
list := corev1.PodList{Items: []corev1.Pod{{
Name: "web-0",
Namespace: "prod",
Annotations: map[string]string{defaultContainer: "sidecar"},
Spec: corev1.PodSpec{Containers: []corev1.Container{{Name: "app"}, {Name: "sidecar"}}},
}}}
w.Header().Set("Content-Type", "application/json")
_ = json.NewEncoder(w).Encode(list)
})
stream, err := Open(t.Context(), client, Request{Namespace: "prod", Selector: "app=web"})
if err != nil {
t.Fatalf("open: %v", err)
}
defer stream.Close()
gather(t, stream, 1)
select {
case container := <-asked:
if container != "sidecar" {
t.Fatalf("container = %q, want the one the pod points at", container)
}
case <-time.After(5 * time.Second):
t.Fatal("no log request was made")
}
}
func TestAChosenContainerBeatsThePodsDefault(t *testing.T) {
asked := make(chan string, 4)
client := clientFor(t, func(w http.ResponseWriter, r *http.Request) {
if strings.HasSuffix(r.URL.Path, "/log") {
asked <- r.URL.Query().Get("container")
_, _ = fmt.Fprintln(w, "hello")
return
}
list := corev1.PodList{Items: []corev1.Pod{{
Name: "web-0",
Namespace: "prod",
Annotations: map[string]string{defaultContainer: "sidecar"},
Spec: corev1.PodSpec{Containers: []corev1.Container{{Name: "app"}, {Name: "sidecar"}}},
}}}
w.Header().Set("Content-Type", "application/json")
_ = json.NewEncoder(w).Encode(list)
})
stream, err := Open(t.Context(), client, Request{
Namespace: "prod",
Selector: "app=web",
Container: "app",
})
if err != nil {
t.Fatalf("open: %v", err)
}
defer stream.Close()
gather(t, stream, 1)
select {
case container := <-asked:
if container != "app" {
t.Fatalf("container = %q, want the one that was asked for", container)
}
case <-time.After(5 * time.Second):
t.Fatal("no log request was made")
}
}
func TestEveryPodRefusingIsTheWholeRequestRefusing(t *testing.T) {
apiserver := newCluster()
apiserver.add("web-0", "hello")
apiserver.add("web-1", "hello")
client := clientFor(t, func(w http.ResponseWriter, r *http.Request) {
if strings.HasSuffix(r.URL.Path, "/log") {
w.WriteHeader(http.StatusForbidden)
_, _ = w.Write([]byte(`{"kind":"Status","message":"pods/log is forbidden","code":403}`))
return
}
apiserver.handler(t)(w, r)
})
_, err := Open(t.Context(), client, Request{Namespace: "prod", Selector: "app=web"})
if err == nil {
t.Fatal("a stream nobody could read reported success and would wait forever")
}
if !strings.Contains(err.Error(), "forbidden") {
t.Fatalf("error = %v, want the apiserver's reason", err)
}
}
func TestAPodThatRefusesDoesNotStopTheOthers(t *testing.T) {
apiserver := newCluster()
apiserver.add("web-0", "hello")
apiserver.add("web-1", "hello")
client := clientFor(t, func(w http.ResponseWriter, r *http.Request) {
if strings.Contains(r.URL.Path, "web-0") && strings.HasSuffix(r.URL.Path, "/log") {
w.WriteHeader(http.StatusForbidden)
return
}
apiserver.handler(t)(w, r)
})
stream, err := Open(t.Context(), client, Request{Namespace: "prod", Selector: "app=web"})
if err != nil {
t.Fatalf("open: %v", err)
}
defer stream.Close()
found := gather(t, stream, 1)
if found["web-1"] == nil {
t.Fatalf("lines = %v, want the pod that did answer", found)
}
if stream.Attached() != 1 {
t.Fatalf("attached = %d, want only the pod actually being read", stream.Attached())
}
if stream.Err() == nil {
t.Fatal("the partial refusal was hidden")
}
}
func TestAMergedStreamReportsAPodWhoseLogsBreakMidRead(t *testing.T) {
apiserver := newCluster()
apiserver.add("web-0", strings.Repeat("x", maxLineBytes+1))
apiserver.add("web-1", "hello")
stream, err := Open(t.Context(), manyClient(t, apiserver), Request{Namespace: "prod", Selector: "app=web"})
if err != nil {
t.Fatalf("open: %v", err)
}
defer stream.Close()
drainLines(t, stream)
if stream.Err() == nil || !strings.Contains(stream.Err().Error(), "web-0") {
t.Fatalf("error = %v, want the broken pod named", stream.Err())
}
}
func TestASinglePodThatIsStillStartingDoesNotFailAFollowingStream(t *testing.T) {
restore := hurryResolve(t)
defer restore()
apiserver := newCluster()
apiserver.add("web-0", "ready now")
var mu sync.Mutex
refused := false
client := clientFor(t, func(w http.ResponseWriter, r *http.Request) {
if strings.HasSuffix(r.URL.Path, "/log") {
mu.Lock()
first := !refused
refused = true
mu.Unlock()
if first {
w.WriteHeader(http.StatusBadRequest)
_, _ = w.Write([]byte(`{"kind":"Status","message":"container is creating","code":400}`))
return
}
}
apiserver.handler(t)(w, r)
})
stream, err := Open(t.Context(), client, Request{
Namespace: "prod",
Selector: "app=web",
Follow: true,
})
if err != nil {
t.Fatalf("open: %v, want the stream to wait for the pod to start", err)
}
defer stream.Close()
found := gather(t, stream, 1)
if found["web-0"] == nil {
t.Fatalf("lines = %v, want the pod once it started", found)
}
}
func TestAStreamThatIsNotFollowingFailsWhenNothingCanBeRead(t *testing.T) {
apiserver := newCluster()
apiserver.add("web-0", "hello")
client := clientFor(t, func(w http.ResponseWriter, r *http.Request) {
if strings.HasSuffix(r.URL.Path, "/log") {
w.WriteHeader(http.StatusBadRequest)
_, _ = w.Write([]byte(`{"kind":"Status","message":"container is creating","code":400}`))
return
}
apiserver.handler(t)(w, r)
})
_, err := Open(t.Context(), client, Request{Namespace: "prod", Selector: "app=web"})
if err == nil {
t.Fatal("a one-shot read of nothing reported success")
}
}
func TestOnePodKeepsTheWholeTail(t *testing.T) {
apiserver := newCluster()
apiserver.add("web-0", "hello")
stream, err := Open(t.Context(), manyClient(t, apiserver), Request{
Namespace: "prod",
Selector: "app=web",
TailLines: 500,
})
if err != nil {
t.Fatalf("open: %v", err)
}
defer stream.Close()
gather(t, stream, 1)
if apiserver.askedForTail() != "500" {
t.Fatalf("tailLines = %q, want the whole tail for a single pod", apiserver.askedForTail())
}
}
func TestManyPodsShareTheTailBetweenThem(t *testing.T) {
apiserver := newCluster()
for i := range maxPods {
apiserver.add(fmt.Sprintf("web-%02d", i), "hello")
}
stream, err := Open(t.Context(), manyClient(t, apiserver), Request{
Namespace: "prod",
Selector: "app=web",
TailLines: 500,
})
if err != nil {
t.Fatalf("open: %v", err)
}
defer stream.Close()
gather(t, stream, 1)
if apiserver.askedForTail() != "250" {
t.Fatalf("tailLines = %q, want the budget split across the pods", apiserver.askedForTail())
}
}
func TestTheTailNeverShrinksBelowSomethingUseful(t *testing.T) {
if share(500, 1000) != minTail {
t.Fatalf("share = %d, want a floor of %d", share(500, 1000), minTail)
}
if share(0, 20) != 0 {
t.Fatalf("share = %d, want no tail when none was asked for", share(0, 20))
}
if share(10, 20) != 10 {
t.Fatalf("share = %d, want a small tail left alone", share(10, 20))
}
}
func (apiserver *cluster) remove(name string) {
apiserver.mu.Lock()
defer apiserver.mu.Unlock()
apiserver.pods = slices.DeleteFunc(apiserver.pods, func(held string) bool {
return held == name
})
}
func waitForAttached(t *testing.T, stream *Stream, want int) {
t.Helper()
deadline := time.After(5 * time.Second)
for stream.Attached() != want {
select {
case <-deadline:
t.Fatalf("the stream is reading %d pods, want %d", stream.Attached(), want)
case <-time.After(time.Millisecond):
}
}
}
func TestAPodThatComesBackIsReadAgain(t *testing.T) {
restore := hurryResolve(t)
defer restore()
apiserver := newCluster()
apiserver.add("web-0", "first time")
stream, err := Open(t.Context(), manyClient(t, apiserver), Request{
Namespace: "prod",
Selector: "app=web",
Follow: true,
})
if err != nil {
t.Fatalf("open: %v", err)
}
defer stream.Close()
gather(t, stream, 1)
apiserver.remove("web-0")
waitForAttached(t, stream, 0)
apiserver.add("web-0", "second time")
found := gather(t, stream, 1)
if found["web-0"] == nil {
t.Fatalf("lines = %v, want the pod read again after it came back", found)
}
}
func TestAPodRecreatedWithoutAListingGapIsReadAgain(t *testing.T) {
restore := hurryResolve(t)
defer restore()
var mu sync.Mutex
generation := "first"
logReads := 0
client := clientFor(t, func(w http.ResponseWriter, r *http.Request) {
mu.Lock()
current := generation
if strings.HasSuffix(r.URL.Path, "/log") {
logReads++
mu.Unlock()
_, _ = fmt.Fprintln(w, current)
return
}
mu.Unlock()
list := corev1.PodList{Items: []corev1.Pod{{
Name: "web-0", Namespace: "prod", UID: types.UID(current),
}}}
w.Header().Set("Content-Type", "application/json")
_ = json.NewEncoder(w).Encode(list)
})
stream, err := Open(t.Context(), client, Request{Namespace: "prod", Selector: "app=web", Follow: true})
if err != nil {
t.Fatalf("open: %v", err)
}
defer stream.Close()
gather(t, stream, 1)
mu.Lock()
generation = "second"
mu.Unlock()
select {
case line := <-stream.Lines:
if line.Text != "second" {
t.Fatalf("line = %q, want the replacement pod", line.Text)
}
case <-time.After(time.Second):
mu.Lock()
reads := logReads
mu.Unlock()
t.Fatalf("the replacement pod was not read; log endpoint was opened %d time(s)", reads)
}
}
func TestTheCapHoldsForPodsThatTurnUpLater(t *testing.T) {
restore := hurryResolve(t)
defer restore()
apiserver := newCluster()
apiserver.holdOpen()
for i := range maxPods {
apiserver.add(fmt.Sprintf("web-%02d", i), "hello")
}
stream, err := Open(t.Context(), manyClient(t, apiserver), Request{
Namespace: "prod",
Selector: "app=web",
Follow: true,
})
if err != nil {
t.Fatalf("open: %v", err)
}
defer stream.Close()
gather(t, stream, 1)
for i := maxPods; i < maxPods+5; i++ {
apiserver.add(fmt.Sprintf("web-%02d", i), "hello")
}
deadline := time.Now().Add(2 * time.Second)
for time.Now().Before(deadline) {
if stream.Matched() > maxPods {
break
}
time.Sleep(10 * time.Millisecond)
}
if stream.Attached() > maxPods {
t.Fatalf("attached = %d, want no more than the cap of %d", stream.Attached(), maxPods)
}
if stream.Matched() != maxPods+5 {
t.Fatalf("matched = %d, want every pod counted", stream.Matched())
}
}
func TestNoErrorIsNotAPermanentRefusal(t *testing.T) {
if permanent(nil) {
t.Fatal("a pod that opened fine was treated as refused")
}
}