repo_description: 🐶 Kubernetes CLI To Manage Your Clusters In Style!
repo_topics: go golang k8s k8s-cluster k9s kubernetes kubernetes-cli kubernetes-clusters
code: // SPDX-License-Identifier: Apache-2.0
  // Copyright Authors of K9s
  
  package dao
  
  import (
  	"bufio"
  	"context"
  	"errors"
  	"fmt"
  	"io"
  	"log/slog"
  	"sync"
  	"time"
  
  	"github.com/cenkalti/backoff/v4"
  	"github.com/derailed/k9s/internal"
  	"github.com/derailed/k9s/internal/client"
  	"github.com/derailed/k9s/internal/render"
  	"github.com/derailed/k9s/internal/slogs"
  	"github.com/derailed/k9s/internal/watch"
  	"github.com/derailed/tview"
  	v1 "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"
  	"k8s.io/apimachinery/pkg/util/sets"
  	restclient "k8s.io/client-go/rest"
  	mv1beta1 "k8s.io/metrics/pkg/apis/metrics/v1beta1"
  )
  
  var (
  	_ Accessor        = (*Pod)(nil)
  	_ Nuker           = (*Pod)(nil)
  	_ Loggable        = (*Pod)(nil)
  	_ Controller      = (*Pod)(nil)
  	_ ContainsPodSpec = (*Pod)(nil)
  	_ ImageLister     = (*Pod)(nil)
  )
  
  type streamResult int
  
  const (
  	logRetryCount                  = 20
  	logBackoffInitial              = 500 * time.Millisecond
  	logBackoffMax                  = 30 * time.Second
  	logChannelBuffer               = 50   // Buffer size for log channel to reduce drops
  	streamEOF         streamResult = iota // legit container log close (no retry)
  	streamError                           // retryable error (network, auth, etc.)
  	streamCanceled                        // context canceled
  )
  
  // Pod represents a pod resource.
  type Pod struct {
  	Resource
  }
  
  // shouldStopRetrying checks if we should stop retrying log streaming based on pod status.
  func (p *Pod) shouldStopRetrying(path string) bool {
  	pod, err := p.GetInstance(path)
  	if err != nil {
  		return true
  	}
  
  	if pod.DeletionTimestamp != nil {
  		return true
  	}
  
  	switch pod.Status.Phase {
  	case v1.PodSucceeded, v1.PodFailed:
  		return true
  	default:
  		return false
  	}
  }
  
  // Get returns a resource instance if found, else an error.
  func (p *Pod) Get(ctx context.Context, path string) (runtime.Object, error) {
  	o, err := p.Resource.Get(ctx, path)
  	if err != nil {
  		return o, err
  	}
  
  	u, ok := o.(*unstructured.Unstructured)
  	if !ok {
  		return nil, fmt.Errorf("expecting *unstructured.Unstructured but got `%T", o)
  	}
  
  	var pmx *mv1beta1.PodMetrics
  	if withMx, ok := ctx.Value(internal.KeyWithMetrics).(bool); ok && withMx {
  		pmx, _ = client.DialMetrics(p.Client()).FetchPodMetrics(ctx, path)
  	}
  
  	return &render.PodWithMetrics{Raw: u, MX: pmx}, nil
  }
  
  // ListImages lists container images.
  func (p *Pod) ListImages(_ context.Context, path string) ([]string, error) {
  	pod, err := p.GetInstance(path)
  	if err != nil {
  		return nil, err
  	}
  
  	return render.ExtractImages(&pod.Spec), nil
  }
  
  // List returns a collection of nodes.
  func (p *Pod) List(ctx context.Context, ns string) ([]runtime.Object, error) {
  	oo, err := p.Resource.List(ctx, ns)
  	if err != nil {
  		return oo, err
  	}
  
  	var pmx client.PodsMetricsMap
  	if withMx, ok := ctx.Value(internal.KeyWithMetrics).(bool); ok && withMx {
  		pmx, _ = client.DialMetrics(p.Client()).FetchPodsMetricsMap(ctx, ns)
  	}
  	sel, _ := ctx.Value(internal.KeyFields).(string)
  	fsel, err := labels.ConvertSelectorToLabelsMap(sel)
  	if err != nil {
  		return nil, err
  	}
  	nodeName := fsel["spec.nodeName"]
  
  	res := make([]runtime.Object, 0, len(oo))
  	for _, o := range oo {
  		u, ok := o.(*unstructured.Unstructured)
  		if !ok {
  			return res, fmt.Errorf("expecting *unstructured.Unstructured but got `%T", o)
  		}
  		fqn := extractFQN(o)
  		if nodeName == "" {
  			res = append(res, &render.PodWithMetrics{Raw: u, MX: pmx[fqn]})
  			continue
  		}
  
  		spec, ok := u.Object["spec"].(map[string]any)
  		if !ok {
  			return res, fmt.Errorf("expecting interface map but got `%T", o)
  		}
  		if spec["nodeName"] == nodeName {
  			res = append(res, &render.PodWithMetrics{Raw: u, MX: pmx[fqn]})
  		}
  	}
  
  	return res, nil
  }
  
  // Logs fetch container logs for a given pod and container.
  func (p *Pod) Logs(path string, opts *v1.PodLogOptions) (*restclient.Request, error) {
  	ns, n := client.Namespaced(path)
  	auth, err := p.Client().CanI(ns, client.NewGVR(client.PodGVR.String()+":log"), n, client.GetAccess)
  	if err != nil {
  		return nil, err
  	}
  	if !auth {
  		return nil, fmt.Errorf("user is not authorized to view pod logs")
  	}
  
  	dial, err := p.Client().DialLogs()
  	if err != nil {
  		return nil, err
  	}
  
  	return dial.CoreV1().Pods(ns).GetLogs(n, opts), nil
  }
  
  // Containers returns all container names on pod.
  func (p *Pod) Containers(path string, includeInit bool) ([]string, error) {
  	pod, err := p.GetInstance(path)
  	if err != nil {
  		return nil, err
  	}
  
  	cc := make([]string, 0, len(pod.Spec.Containers)+len(pod.Spec.InitContainers))
  	for i := range pod.Spec.Containers {
  		cc = append(cc, pod.Spec.Containers[i].Name)
  	}
  
  	if includeInit {
  		for i := range pod.Spec.InitContainers {
  			cc = append(cc, pod.Spec.InitContainers[i].Name)
  		}
  	}
  
  	return cc, nil
  }
  
  // Pod returns a pod victim by name.
  func (*Pod) Pod(fqn string) (string, error) {
  	return fqn, nil
  }
  
  // GetInstance returns a pod instance.
  func (p *Pod) GetInstance(fqn string) (*v1.Pod, error) {
  	o, err := p.getFactory().Get(p.gvr, fqn, true, labels.Everything())
  	if err != nil {
  		return nil, err
  	}
  
  	var pod v1.Pod
  	err = runtime.DefaultUnstructuredConverter.FromUnstructured(o.(*unstructured.Unstructured).Object, &pod)
  	if err != nil {
  		return nil, err
  	}
  
  	return &pod, nil
  }
  
  // TailLogs tails a given container logs.
  func (p *Pod) TailLogs(ctx context.Context, opts *LogOptions) ([]LogChan, error) {
  	fac, ok := ctx.Value(internal.KeyFactory).(*watch.Factory)
  	if !ok {
  		return nil, errors.New("no factory in context")
  	}
  	o, err := fac.Get(p.gvr, opts.Path, true, labels.Everything())
  	if err != nil {
  		return nil, err
  	}
  	var po v1.Pod
  	if err := runtime.DefaultUnstructuredConverter.FromUnstructured(o.(*unstructured.Unstructured).Object, &po); err != nil {
  		return nil, err
  	}
  	coCounts := len(po.Spec.InitContainers) + len(po.Spec.Containers) + len(po.Spec.EphemeralContainers)
  	if coCounts == 1 {
  		opts.SingleContainer = true
  	}
  
  	outs := make([]LogChan, 0, coCounts)
  	if co, ok := GetDefaultContainer(&po.ObjectMeta, &po.Spec); ok && !opts.AllContainers {
  		opts.DefaultContainer = co
  		return append(outs, tailLogs(ctx, p, opts)), nil
  	}
  	if opts.HasContainer() && !opts.AllContainers {
  		return append(outs, tailLogs(ctx, p, opts)), nil
  	}
  	for i := range po.Spec.InitContainers {
  		cfg := opts.Clone()
  		cfg.Container = po.Spec.InitContainers[i].Name
  		outs = append(outs, tailLogs(ctx, p, cfg))
  	}
  	for i := range po.Spec.Containers {
  		cfg := opts.Clone()
  		cfg.Container = po.Spec.Containers[i].Name
  		outs = append(outs, tailLogs(ctx, p, cfg))
  	}
  	for i := range po.Spec.EphemeralContainers {
  		cfg := opts.Clone()
  		cfg.Container = po.Spec.EphemeralContainers[i].Name
  		outs = append(outs, tailLogs(ctx, p, cfg))
  	}
  
  	return outs, nil
  }
  
  // ScanSA scans for ServiceAccount refs.
  func (p *Pod) ScanSA(_ context.Context, fqn string, wait bool) (Refs, error) {
  	ns, n := client.Namespaced(fqn)
  	oo, err := p.getFactory().List(p.gvr, ns, wait, labels.Everything())
  	if err != nil {
  		return nil, err
  	}
  
  	refs := make(Refs, 0, len(oo))
  	for _, o := range oo {
  		var pod v1.Pod
  		err = runtime.DefaultUnstructuredConverter.FromUnstructured(o.(*unstructured.Unstructured).Object, &pod)
  		if err != nil {
  			return nil, errors.New("expecting Deployment resource")
  		}
  		// Just pick controller less pods...
  		if len(pod.OwnerReferences) > 0 {
  			continue
  		}
  		if serviceAccountMatches(pod.Spec.ServiceAccountName, n) {
  			refs = append(refs, Ref{
  				GVR: p.GVR(),
  				FQN: client.FQN(pod.Namespace, pod.Name),
  			})
  		}
  	}
  
  	return refs, nil
  }
  
  // Scan scans for cluster resource refs.
  func (p *Pod) Scan(_ context.Context, gvr *client.GVR, fqn string, wait bool) (Refs, error) {
  	ns, n := client.Namespaced(fqn)
  	oo, err := p.getFactory().List(p.gvr, ns, wait, labels.Everything())
  	if err != nil {
  		return nil, err
  	}
  
  	refs := make(Refs, 0, len(oo))
  	for _, o := range oo {
  		var pod v1.Pod
  		err = runtime.DefaultUnstructuredConverter.FromUnstructured(o.(*unstructured.Unstructured).Object, &pod)
  		if err != nil {
  			return nil, errors.New("expecting Pod resource")
  		}
  		// Just pick controller less pods...
  		if len(pod.OwnerReferences) > 0 {
  			continue
  		}
  		switch gvr {
  		case client.CmGVR:
  			if !hasConfigMap(&pod.Spec, n) {
  				continue
  			}
  			refs = append(refs, Ref{
  				GVR: p.GVR(),
  				FQN: client.FQN(pod.Namespace, pod.Name),
  			})
  		case client.SecGVR:
  			found, err := hasSecret(p.Factory, &pod.Spec, pod.Namespace, n, wait)
  			if err != nil {
  				slog.Warn("Locate secret failed",
  					slogs.FQN, fqn,
  					slogs.Error, err,
  				)
  				continue
  			}
  			if !found {
  				continue
  			}
  			refs = append(refs, Ref{
  				GVR: p.GVR(),
  				FQN: client.FQN(pod.Namespace, pod.Name),
  			})
  		case client.PvcGVR:
  			if !hasPVC(&pod.Spec, n) {
  				continue
  			}
  			refs = append(refs, Ref{
  				GVR: p.GVR(),
  				FQN: client.FQN(pod.Namespace, pod.Name),
  			})
  		case client.PcGVR:
  			if !hasPC(&pod.Spec, n) {
  				continue
  			}
  			refs = append(refs, Ref{
  				GVR: p.GVR(),
  				FQN: client.FQN(pod.Namespace, pod.Name),
  			})
  		}
  	}
  
  	return refs, nil
  }
  
  // ----------------------------------------------------------------------------
  // Helpers...
  
  func tailLogs(ctx context.Context, logger Logger, opts *LogOptions) LogChan {
  	out := make(LogChan, logChannelBuffer)
  	var wg sync.WaitGroup
  
  	wg.Add(1)
  	go func() {
  		defer wg.Done()
  		podOpts := opts.ToPodLogOptions()
  
  		// Setup exponential backoff following project pattern
  		bf := backoff.NewExponentialBackOff()
  		bf.InitialInterval = logBackoffInitial
  		bf.MaxElapsedTime = 0
  		bf.MaxInterval = logBackoffMax / 2
  		backoffCtx := backoff.WithContext(bf, ctx)
  		delay := logBackoffInitial
  
  		for range logRetryCount {
  			req, err := logger.Logs(opts.Path, podOpts)
  			if err != nil {
  				slog.Error("Log request failed",
  					slogs.Container, opts.Info(),
  					slogs.Error, err,
  				)
  				// Check if we should stop retrying based on pod status
  				if pod, ok := logger.(*Pod); ok && pod.shouldStopRetrying(opts.Path) {
  					slog.Debug("Stopping log retry - pod is terminating or deleted",
  						slogs.Container, opts.Info(),
  					)
  					return
  				}
  				select {
  				case <-ctx.Done():
  					return
  				case <-time.After(delay):
  					if delay = backoffCtx.NextBackOff(); delay == backoff.Stop {
  						return
  					}
  				}
  				continue
  			}
  
  			stream, e := req.Stream(ctx)
  			if e != nil {
  				slog.Error("Stream logs failed",
  					slogs.Error, e,
  					slogs.Container, opts.Info(),
  				)
  				// Check if we should stop retrying based on pod status
  				if pod, ok := logger.(*Pod); ok && pod.shouldStopRetrying(opts.Path) {
  					slog.Debug("Stopping log retry - pod is terminating or deleted",
  						slogs.Container, opts.Info(),
  					)
  					return
  				}
  				select {
  				case <-ctx.Done():
  					return
  				case <-time.After(delay):
  					if delay = backoffCtx.NextBackOff(); delay == backoff.Stop {
  						return
  					}
  				}
  				continue
  			}
  
  			// Process logs until completion
  			result := readLogs(ctx, stream, out, opts)
  			switch result {
  			case streamEOF:
  				slog.Debug("Log stream ended cleanly",
  					slogs.Container, opts.Info(),
  				)
  				return
  			case streamError:
  				// Check if we should stop retrying based on pod status
  				if pod, ok := logger.(*Pod); ok && pod.shouldStopRetrying(opts.Path) {
  					slog.Debug("Stopping log retry after stream error - pod is terminating or deleted",
  						slogs.Container, opts.Info(),
  					)
  					return
  				}
  				slog.Debug("Log stream error, retrying",
  					slogs.Container, opts.Info(),
  				)
  				select {
  				case <-ctx.Done():
  					return
  				case <-time.After(delay):
  					if delay = backoffCtx.NextBackOff(); delay == backoff.Stop {
  						return
  					}
  				}
  				continue
  			case streamCanceled:
  				return
  			}
  
  			// Reset backoff and delay on successful connection
  			bf.Reset()
  			delay = logBackoffInitial
  		}
  
  		// Out of retries
  		out <- opts.ToErrLogItem(fmt.Errorf("failed to maintain log stream after %d retries", logRetryCount))
  	}()
  
  	go func() {
  		wg.Wait()
  		close(out)
  	}()
  
  	return out
  }
  
  func readLogs(ctx context.Context, stream io.ReadCloser, out chan<- *LogItem, opts *LogOptions) streamResult {
  	defer func() {
  		if err := stream.Close(); err != nil && !errors.Is(err, io.ErrClosedPipe) {
  			slog.Error("Failed to close stream",
  				slogs.Container, opts.Info(),
  				slogs.Error, err,
  			)
  		}
  	}()
  
  	r := bufio.NewReader(stream)
  
  	for {
  		bytes, err := r.ReadBytes('\n')
  		if err == nil {
  			item := opts.ToLogItem(tview.EscapeBytes(bytes))
  			select {
  			case <-ctx.Done():
  				return streamCanceled
  			case out <- item:
  			default:
  				// Avoid deadlock if consumer is too slow
  				slog.Warn("Dropping log line due to slow consumer",
  					slogs.Container, opts.Info(),
  				)
  			}
  			continue
  		}
  
  		if errors.Is(err, io.EOF) {
  			if len(bytes) > 0 {
  				// Emit trailing partial line before EOF
  				out <- opts.ToLogItem(tview.EscapeBytes(bytes))
  			}
  			slog.Debug("Log reader reached EOF", slogs.Container, opts.Info())
  			out <- opts.ToErrLogItem(fmt.Errorf("stream closed: %w for %s", err, opts.Info()))
  			return streamEOF
  		}
  
  		// Non-EOF error
  		slog.Debug("Log stream error, will retry connection",
  			slogs.Container, opts.Info(),
  			slogs.Error, fmt.Errorf("stream error: %w for %s", err, opts.Info()),
  		)
  		// Don't send stream errors to user - they will be retried
  		// Only final retry exhaustion message is shown
  		return streamError
  	}
  }
  
  // MetaFQN returns a fully qualified resource name.
  func MetaFQN(m *metav1.ObjectMeta) string {
  	if m.Namespace == "" {
  		return m.Name
  	}
  
  	return FQN(m.Namespace, m.Name)
  }
  
  // GetPodSpec returns a pod spec given a resource.
  func (p *Pod) GetPodSpec(path string) (*v1.PodSpec, error) {
  	pod, err := p.GetInstance(path)
  	if err != nil {
  		return nil, err
  	}
  	podSpec := pod.Spec
  
  	return &podSpec, nil
  }
  
  // SetImages sets container images.
  func (p *Pod) SetImages(ctx context.Context, path string, imageSpecs ImageSpecs) error {
  	ns, n := client.Namespaced(path)
  	auth, err := p.Client().CanI(ns, p.gvr, n, client.PatchAccess)
  	if err != nil {
  		return err
  	}
  	if !auth {
  		return fmt.Errorf("user is not authorized to patch a deployment")
  	}
  	manager, isManaged, err := p.isControlled(path)
  	if err != nil {
  		return err
  	}
  	if isManaged {
  		return fmt.Errorf("unable to set image. This pod is managed by %s. Please set the image on the controller", manager)
  	}
  	jsonPatch, err := GetJsonPatch(imageSpecs)
  	if err != nil {
  		return err
  	}
  	dial, err := p.Client().Dial()
  	if err != nil {
  		return err
  	}
  	_, err = dial.CoreV1().Pods(ns).Patch(
  		ctx,
  		n,
  		types.StrategicMergePatchType,
  		jsonPatch,
  		metav1.PatchOptions{},
  	)
  
  	return err
  }
  
  func (p *Pod) isControlled(path string) (fqn string, ok bool, err error) {
  	pod, err := p.GetInstance(path)
  	if err != nil {
  		return "", false, err
  	}
  	references := pod.GetObjectMeta().GetOwnerReferences()
  	if len(references) > 0 {
  		return fmt.Sprintf("%s/%s", references[0].Kind, references[0].Name), true, nil
  	}
  
  	return "", false, nil
  }
  
  var toastPhases = sets.New(
  	render.PhaseCompleted,
  	render.PhasePending,
  	render.PhaseCrashLoop,
  	render.PhaseError,
  	render.PhaseImagePullBackOff,
  	render.PhaseContainerStatusUnknown,
  	render.PhaseEvicted,
  	render.PhaseOOMKilled,
  )
  
  func (p *Pod) Sanitize(ctx context.Context, ns string) (int, error) {
  	oo, err := p.Resource.List(ctx, ns)
  	if err != nil {
  		return 0, err
  	}
  
  	var count int
  	for _, o := range oo {
  		u, ok := o.(*unstructured.Unstructured)
  		if !ok {
  			continue
  		}
  		var pod v1.Pod
  		err = runtime.DefaultUnstructuredConverter.FromUnstructured(u.Object, &pod)
  		if err != nil {
  			continue
  		}
  
  		if toastPhases.Has(render.PodStatus(&pod)) {
  			// !!BOZO!! Might need to bump timeout otherwise rev limit if too many??
  			fqn := client.FQN(pod.Namespace, pod.Name)
  			slog.Debug("Sanitizing resource", slogs.FQN, fqn)
  			if err := p.Delete(ctx, fqn, nil, 0); err != nil {
  				slog.Debug("Aborted! Sanitizer delete failed",
  					slogs.FQN, fqn,
  					slogs.Count, count,
  					slogs.Error, err,
  				)
  				return count, err
  			}
  			count++
  		}
  	}
  	slog.Debug("Sanitizer deleted pods", slogs.Count, count)
  
  	return count, nil
  }
  