kubectl-persistent-logger/logs/watcher.go

154 lines
3.7 KiB
Go
Raw Normal View History

2022-05-29 19:04:02 +00:00
package logs
import (
"context"
2022-05-31 04:11:44 +00:00
"errors"
2022-05-29 19:04:02 +00:00
"io"
"log"
2022-05-31 04:11:44 +00:00
"sync"
2022-05-29 19:04:02 +00:00
"time"
2022-06-01 17:19:55 +00:00
appsv1 "k8s.io/api/apps/v1"
2022-05-29 19:04:02 +00:00
corev1 "k8s.io/api/core/v1"
2022-06-01 17:19:55 +00:00
apierrors "k8s.io/apimachinery/pkg/api/errors"
2022-05-29 19:04:02 +00:00
metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
"k8s.io/apimachinery/pkg/watch"
"k8s.io/client-go/kubernetes"
)
2022-05-31 04:11:44 +00:00
// KubernetesClient wraps a Kubernetes clientset.
2022-05-30 10:10:58 +00:00
type KubernetesClient struct {
kubernetes.Interface
}
2022-06-01 17:19:55 +00:00
// concurrentWriter implements an io.Writer that can be safely written to from
// multiple goroutines.
2022-05-31 04:11:44 +00:00
type concurrentWriter struct {
w io.Writer
mu sync.Mutex
}
// Write implements io.Writer.
func (cw *concurrentWriter) Write(p []byte) (int, error) {
cw.mu.Lock()
defer cw.mu.Unlock()
return cw.w.Write(p)
}
2022-06-01 17:19:55 +00:00
type PodWatcherInterface interface {
WatchPods(ctx context.Context) error
Close()
2022-05-31 04:11:44 +00:00
}
2022-06-01 17:19:55 +00:00
// PodWatcherFunc builds a PodWatcher.
type PodWatcherFunc func(KubernetesClient, string, *metav1.LabelSelector, io.Writer) PodWatcherInterface
2022-05-31 04:11:44 +00:00
2022-05-29 19:04:02 +00:00
// Watcher watches a deployment and tails the logs for its currently active
// pods.
type Watcher struct {
2022-06-01 20:04:20 +00:00
deployName string
container string
strictExist bool
clientset KubernetesClient
deployment *appsv1.Deployment
podWatcher PodWatcherInterface
podWatcherFunc PodWatcherFunc
errChan chan error
dst *concurrentWriter
2022-05-29 19:04:02 +00:00
}
// NewWatcher creates a new Watcher.
2022-06-01 20:04:20 +00:00
func NewWatcher(deployName string, container string, strictExist bool, clientset KubernetesClient, podWatcherFunc PodWatcherFunc, dst io.Writer) *Watcher {
2022-05-29 19:04:02 +00:00
return &Watcher{
2022-06-01 20:04:20 +00:00
deployName: deployName,
container: container,
strictExist: strictExist,
clientset: clientset,
podWatcherFunc: podWatcherFunc,
errChan: make(chan error),
dst: &concurrentWriter{w: dst},
2022-05-29 19:04:02 +00:00
}
}
// Watch watches a deployment.
func (w *Watcher) Watch(ctx context.Context) error {
deploymentsClient := w.clientset.AppsV1().Deployments(corev1.NamespaceDefault)
2022-06-01 17:19:55 +00:00
// Check if the deployment exists before we commence watching, to allow us to
// return an error if needed.
_, err := deploymentsClient.Get(ctx, w.deployName, metav1.GetOptions{})
2022-06-01 20:04:02 +00:00
var statusErr *apierrors.StatusError
if errors.As(err, &statusErr) && statusErr.Status().Reason == metav1.StatusReasonNotFound {
if w.strictExist {
2022-06-01 17:19:55 +00:00
return err
}
log.Printf(`deployment "%s" does not exist, waiting`, w.deployName)
2022-05-29 19:04:02 +00:00
}
2022-06-01 17:19:55 +00:00
opts := metav1.ListOptions{Watch: true, FieldSelector: "metadata.name=" + w.deployName}
deploymentsWatcher, err := deploymentsClient.Watch(ctx, opts)
2022-05-29 19:04:02 +00:00
if err != nil {
return err
}
2022-06-01 17:19:55 +00:00
defer deploymentsWatcher.Stop()
2022-05-29 19:04:02 +00:00
2022-06-01 17:19:55 +00:00
ticker := time.NewTicker(time.Second)
defer ticker.Stop()
2022-05-29 19:04:02 +00:00
2022-06-04 00:27:20 +00:00
resultChan := deploymentsWatcher.ResultChan()
2022-05-29 19:04:02 +00:00
for {
select {
2022-06-04 00:27:20 +00:00
case evt, ok := <-resultChan:
if !ok {
resultChan = nil
continue
}
2022-05-29 19:04:02 +00:00
switch evt.Type {
case watch.Added, watch.Modified:
2022-06-01 17:19:55 +00:00
deployment := evt.Object.(*appsv1.Deployment)
w.addDeployment(ctx, deployment)
2022-05-29 19:04:02 +00:00
case watch.Deleted:
2022-06-01 17:19:55 +00:00
w.removeDeployment()
2022-05-29 19:04:02 +00:00
}
2022-06-01 17:19:55 +00:00
case err := <-w.errChan:
return err
case <-ctx.Done():
return ctx.Err()
2022-05-29 19:04:02 +00:00
}
}
}
2022-06-01 17:19:55 +00:00
func (w *Watcher) addDeployment(ctx context.Context, deployment *appsv1.Deployment) {
if w.deployment != nil && w.deployment.UID == deployment.UID {
return
2022-05-31 04:11:44 +00:00
}
2022-06-01 17:19:55 +00:00
w.removeDeployment()
2022-05-31 04:11:44 +00:00
2022-06-01 17:19:55 +00:00
log.Println("[DeploymentWatcher] add deployment")
2022-05-31 04:11:44 +00:00
2022-06-01 17:19:55 +00:00
w.deployment = deployment
w.podWatcher = w.podWatcherFunc(
w.clientset,
w.container,
deployment.Spec.Selector,
w.dst,
)
2022-05-31 04:11:44 +00:00
2022-06-01 17:19:55 +00:00
go func() {
if err := w.podWatcher.WatchPods(ctx); err != nil {
w.errChan <- err
}
}()
2022-05-29 19:04:02 +00:00
}
2022-06-01 17:19:55 +00:00
func (w *Watcher) removeDeployment() {
if w.podWatcher != nil {
log.Println("[DeploymentWatcher] remove deployment")
w.podWatcher.Close()
w.podWatcher = nil
}
w.deployment = nil
2022-05-29 19:04:02 +00:00
}