kubectl-persistent-logger/logs/watcher_test.go

156 lines
5.4 KiB
Go
Raw Normal View History

2022-05-30 10:10:58 +00:00
package logs_test
import (
"bytes"
"context"
2022-06-01 17:19:55 +00:00
"errors"
"io"
2022-06-10 17:04:52 +00:00
"log"
2022-06-01 17:19:55 +00:00
"strings"
2022-05-30 10:10:58 +00:00
"testing"
"time"
"git.netflux.io/rob/kubectl-persistent-logger/logs"
"github.com/stretchr/testify/assert"
"github.com/stretchr/testify/require"
corev1 "k8s.io/api/core/v1"
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/apis/meta/v1/unstructured"
"k8s.io/apimachinery/pkg/labels"
"k8s.io/apimachinery/pkg/runtime"
"k8s.io/apimachinery/pkg/types"
2022-05-30 10:10:58 +00:00
"k8s.io/apimachinery/pkg/watch"
dynamicclient "k8s.io/client-go/dynamic/fake"
2022-05-30 10:10:58 +00:00
testclient "k8s.io/client-go/kubernetes/fake"
k8stest "k8s.io/client-go/testing"
)
2022-06-01 17:19:55 +00:00
type mockPodWatcher struct{ err error }
func (m *mockPodWatcher) WatchPods(ctx context.Context) error { return m.err }
func (m *mockPodWatcher) Close() {}
func mockPodwatcherFunc(err error) logs.PodWatcherFunc {
2022-06-14 05:47:38 +00:00
return func(logs.KubernetesClient, string, string, labels.Selector, io.Writer, *log.Logger) logs.PodWatcherInterface {
2022-06-01 17:19:55 +00:00
return &mockPodWatcher{err: err}
}
}
2022-06-10 17:04:52 +00:00
func discardLogger() *log.Logger {
return log.New(io.Discard, "", 0)
}
func buildDeployment(t *testing.T, name string) *unstructured.Unstructured {
deployment := new(unstructured.Unstructured)
deployment.SetAPIVersion("v1")
deployment.SetKind("deployment")
deployment.SetName("mydeployment")
deployment.SetNamespace("default")
deployment.SetUID(types.UID("foo"))
require.NoError(t, unstructured.SetNestedField(deployment.Object, map[string]any{"app": "myapp"}, "spec", "selector", "matchLabels"))
return deployment
}
func TestWatcherStrictExist(t *testing.T) {
client := logs.KubernetesClient{Untyped: dynamicclient.NewSimpleDynamicClient(runtime.NewScheme())}
2022-06-01 17:19:55 +00:00
var buf bytes.Buffer
params := logs.WatcherParams{Name: "mydeployment", Type: "deployments", Namespace: "default", StrictExist: true}
2022-06-10 17:04:52 +00:00
watcher := logs.NewWatcher(params, client, mockPodwatcherFunc(nil), &buf, discardLogger())
2022-06-01 17:19:55 +00:00
err := watcher.Watch(context.Background())
assert.EqualError(t, err, `deployments.apps "mydeployment" not found`)
}
func TestWatcherPodWatcherError(t *testing.T) {
deploymentsWatcher := watch.NewFake()
untypedClient := dynamicclient.NewSimpleDynamicClient(runtime.NewScheme())
untypedClient.PrependWatchReactor("deployments", k8stest.DefaultWatchReactor(deploymentsWatcher, nil))
client := logs.KubernetesClient{Untyped: untypedClient}
2022-05-30 10:10:58 +00:00
2022-06-01 17:19:55 +00:00
var buf bytes.Buffer
wantErr := errors.New("foo")
params := logs.WatcherParams{Name: "mydeployment", Type: "deployments", Namespace: "default"}
2022-06-10 17:04:52 +00:00
watcher := logs.NewWatcher(params, client, mockPodwatcherFunc(wantErr), &buf, discardLogger())
2022-05-30 10:10:58 +00:00
go func() {
2022-06-01 17:19:55 +00:00
defer deploymentsWatcher.Stop()
deployment := buildDeployment(t, "mydeployment")
2022-06-01 17:19:55 +00:00
deploymentsWatcher.Add(deployment)
}()
err := watcher.Watch(context.Background())
assert.EqualError(t, err, wantErr.Error())
}
2022-06-04 00:27:20 +00:00
func TestWatcherClosedChannel(t *testing.T) {
deploymentsWatcher := watch.NewFake()
untypedClient := dynamicclient.NewSimpleDynamicClient(runtime.NewScheme())
untypedClient.PrependWatchReactor("deployments", k8stest.DefaultWatchReactor(deploymentsWatcher, nil))
client := logs.KubernetesClient{Untyped: untypedClient}
2022-06-04 00:27:20 +00:00
var buf bytes.Buffer
params := logs.WatcherParams{Name: "mydeployment", Type: "deployments", Namespace: "default"}
2022-06-10 17:04:52 +00:00
watcher := logs.NewWatcher(params, client, nil, &buf, discardLogger())
// Immediately stop the watcher, which closes the ResultChan.
// This should be expected to be handled.
deploymentsWatcher.Stop()
2022-06-04 00:27:20 +00:00
ctx, cancel := context.WithTimeout(context.Background(), time.Millisecond*500)
defer cancel()
err := watcher.Watch(ctx)
require.Equal(t, context.DeadlineExceeded, err)
assert.Equal(t, "", buf.String())
}
2022-06-01 17:19:55 +00:00
func TestWatcherWithPodWatcher(t *testing.T) {
2022-06-07 19:14:07 +00:00
ctx, cancel := context.WithTimeout(context.Background(), time.Second*2)
defer cancel()
2022-06-01 17:19:55 +00:00
deploymentsWatcher := watch.NewFake()
2022-06-02 17:23:47 +00:00
defer deploymentsWatcher.Stop()
2022-06-01 17:19:55 +00:00
podsWatcher := watch.NewFake()
2022-06-02 17:23:47 +00:00
defer podsWatcher.Stop()
typedClient := testclient.NewSimpleClientset()
typedClient.PrependWatchReactor("pods", k8stest.DefaultWatchReactor(podsWatcher, nil))
untypedClient := dynamicclient.NewSimpleDynamicClient(runtime.NewScheme())
untypedClient.PrependWatchReactor("deployments", k8stest.DefaultWatchReactor(deploymentsWatcher, nil))
client := logs.KubernetesClient{Typed: typedClient, Untyped: untypedClient}
2022-06-01 17:19:55 +00:00
go func() {
deployment := buildDeployment(t, "mydeployment")
2022-06-01 17:19:55 +00:00
deploymentsWatcher.Add(deployment)
time.Sleep(time.Millisecond * 250)
pods := []*corev1.Pod{
{ObjectMeta: metav1.ObjectMeta{Name: "foo", Namespace: "default"}, Status: corev1.PodStatus{Phase: corev1.PodRunning}},
{ObjectMeta: metav1.ObjectMeta{Name: "bar", Namespace: "default"}, Status: corev1.PodStatus{Phase: corev1.PodRunning}},
{ObjectMeta: metav1.ObjectMeta{Name: "baz", Namespace: "default"}, Status: corev1.PodStatus{Phase: corev1.PodPending}},
2022-06-01 17:19:55 +00:00
}
2022-05-30 10:10:58 +00:00
for _, pod := range pods {
2022-06-01 17:19:55 +00:00
podsWatcher.Add(pod)
time.Sleep(time.Millisecond * 250)
2022-05-30 10:10:58 +00:00
}
2022-06-01 17:19:55 +00:00
}()
2022-05-30 10:10:58 +00:00
var buf bytes.Buffer
params := logs.WatcherParams{Name: "mydeployment", Type: "deployments", Namespace: "default"}
2022-06-10 17:04:52 +00:00
watcher := logs.NewWatcher(params, client, logs.NewPodWatcher, &buf, discardLogger())
2022-05-30 10:10:58 +00:00
err := watcher.Watch(ctx)
2022-06-07 19:14:07 +00:00
require.EqualError(t, err, context.DeadlineExceeded.Error())
lines := bufToLines(&buf)
require.Len(t, lines, 2)
assert.ElementsMatch(t, []string{"[foo] fake logs", "[bar] fake logs"}, lines)
2022-06-01 17:19:55 +00:00
}
2022-06-07 19:14:07 +00:00
func bufToLines(buf *bytes.Buffer) []string {
2022-06-01 17:19:55 +00:00
return strings.Split(strings.TrimSpace(buf.String()), "\n")
2022-05-30 10:10:58 +00:00
}