feat: show a message when the deployment is pending (#553)

* feat: Show a message when the deployment is pending

* chore: updated the max pending time

* chore: update loggers to be consistent with the rest from main

pkg/instance/execution.go

@@ -73,10 +73,7 @@ func (e *execution) StartWithCallback(ctx context.Context, callback func()) erro
 	go func() {
 		err := e.WaitInstanceIsRunning(ctx)
 		if err != nil {
-			e.instance.Logger.WithFields(logrus.Fields{
-				"instance": e.instance.k8sName,
-				"error":    err,
-			}).Errorf("waiting for instance to be running")
+			e.instance.Logger.WithError(err).WithField("instance", e.instance.k8sName).Error("waiting for instance to be running")
 			return
 		}
 		callback()

pkg/instance/network.go

@@ -83,14 +83,15 @@ func (n *network) PortForwardTCP(ctx context.Context, port int) (int, error) {
 		if attempt == maxRetries {
 			return -1, ErrForwardingPort.WithParams(maxRetries)
 		}
-		n.instance.Logger.WithFields(logrus.Fields{
-			"instance":       n.instance.name,
-			"port":           port,
-			"error":          err,
-			"attempt":        attempt,
-			"max":            maxRetries,
-			"retry_interval": retryInterval.String(),
-		}).Debug("forwarding port failed, retrying")
+		n.instance.Logger.
+			WithError(err).
+			WithFields(logrus.Fields{
+				"instance":       n.instance.name,
+				"port":           port,
+				"attempt":        attempt,
+				"max":            maxRetries,
+				"retry_interval": retryInterval.String(),
+			}).Debug("forwarding port failed, retrying")
 	}
 	return localPort, nil
 }
@@ -295,21 +296,15 @@ func (n *network) deployOrPatchService(ctx context.Context, portsTCP, portsUDP [
 func (n *network) enableIfDisabled(ctx context.Context) error {
 	disableNetwork, err := n.IsDisabled(ctx)
 	if err != nil {
-		n.instance.Logger.WithFields(logrus.Fields{
-			"instance": n.instance.name,
-			"error":    err,
-		}).Error("error checking network status for instance")
+		n.instance.Logger.WithError(err).WithField("instance", n.instance.k8sName).Error("error checking network status for instance")
 		return ErrCheckingNetworkStatusForInstance.WithParams(n.instance.k8sName).Wrap(err)
 	}
 
 	if !disableNetwork {
 		return nil
 	}
 	if err := n.Enable(ctx); err != nil {
-		n.instance.Logger.WithFields(logrus.Fields{
-			"instance": n.instance.name,
-			"error":    err,
-		}).Error("error enabling network for instance")
+		n.instance.Logger.WithError(err).WithField("instance", n.instance.k8sName).Error("error enabling network for instance")
 		return ErrEnablingNetworkForInstance.WithParams(n.instance.k8sName).Wrap(err)
 	}
 	return nil

pkg/k8s/custom_resource.go

@@ -43,7 +43,7 @@ func (c *Client) CreateCustomResource(
 		return ErrCreatingCustomResource.WithParams(gvr.Resource).Wrap(err)
 	}
 
-	c.logger.Debugf("CustomResource %s created", name)
+	c.logger.WithField("name", name).Debug("customResource created")
 	return nil
 }
 

pkg/k8s/daemonset.go

@@ -3,6 +3,7 @@ package k8s
 import (
 	"context"
 
+	"github.com/sirupsen/logrus"
 	appv1 "k8s.io/api/apps/v1"
 	v1 "k8s.io/api/core/v1"
 	apierrs "k8s.io/apimachinery/pkg/api/errors"
@@ -49,7 +50,10 @@ func (c *Client) CreateDaemonSet(
 	if err != nil {
 		return nil, ErrCreatingDaemonset.WithParams(name).Wrap(err)
 	}
-	c.logger.Debugf("DaemonSet %s created in namespace %s", name, c.namespace)
+	c.logger.WithFields(logrus.Fields{
+		"name":      name,
+		"namespace": c.namespace,
+	}).Debug("daemonSet created")
 	return created, nil
 }
 
@@ -73,7 +77,10 @@ func (c *Client) UpdateDaemonSet(ctx context.Context,
 	if err != nil {
 		return nil, ErrUpdatingDaemonset.WithParams(name).Wrap(err)
 	}
-	c.logger.Debugf("DaemonSet %s updated in namespace %s", name, c.namespace)
+	c.logger.WithFields(logrus.Fields{
+		"name":      name,
+		"namespace": c.namespace,
+	}).Debug("daemonSet updated")
 	return updated, nil
 }
 
@@ -82,7 +89,10 @@ func (c *Client) DeleteDaemonSet(ctx context.Context, name string) error {
 	if err != nil {
 		return ErrDeletingDaemonset.WithParams(name).Wrap(err)
 	}
-	c.logger.Debugf("DaemonSet %s deleted in namespace %s", name, c.namespace)
+	c.logger.WithFields(logrus.Fields{
+		"name":      name,
+		"namespace": c.namespace,
+	}).Debug("daemonSet deleted")
 	return nil
 }
 

pkg/k8s/errors.go

@@ -133,4 +133,6 @@ var (
 	ErrInvalidServiceAccountName       = errors.New("InvalidServiceAccountName", "invalid service account name %s: %v")
 	ErrInvalidClusterRoleBindingName   = errors.New("InvalidClusterRoleBindingName", "invalid cluster role binding name %s: %v")
 	ErrInvalidServiceName              = errors.New("InvalidServiceName", "invalid service name %s: %v")
+	ErrListingPods                     = errors.New("ListingPods", "failed to list pods")
+	ErrGetPodStatus                    = errors.New("GetPodStatus", "failed to get pod status for pod %s")
 )

pkg/k8s/k8s.go

@@ -29,6 +29,9 @@ const (
 
 	// retryInterval is the interval to wait between retries
 	retryInterval = 100 * time.Millisecond
+
+	// if any pod is pending for more than this duration, a warning is logged
+	defaultMaxPendingDuration = 60 * time.Second
 )
 
 type Client struct {
@@ -37,6 +40,8 @@ type Client struct {
 	dynamicClient   dynamic.Interface
 	namespace       string
 	logger          *logrus.Logger
+	// max duration for any pod to be in pending state, otherwise it triggers a notice to be shown
+	maxPendingDuration time.Duration
 }
 
 var _ KubeManager = &Client{}
@@ -77,16 +82,18 @@ func NewClientCustom(
 	logger *logrus.Logger,
 ) (*Client, error) {
 	kc := &Client{
-		clientset:       cs,
-		discoveryClient: dc,
-		dynamicClient:   dC,
-		namespace:       namespace,
-		logger:          logger,
+		clientset:          cs,
+		discoveryClient:    dc,
+		dynamicClient:      dC,
+		namespace:          namespace,
+		logger:             logger,
+		maxPendingDuration: defaultMaxPendingDuration,
 	}
 	kc.namespace = SanitizeName(namespace)
 	if err := kc.CreateNamespace(ctx, kc.namespace); err != nil {
 		return nil, ErrCreatingNamespace.WithParams(kc.namespace).Wrap(err)
 	}
+	kc.startPendingPodsWarningMonitor(ctx)
 	return kc, nil
 }
 
@@ -105,3 +112,7 @@ func (c *Client) Namespace() string {
 func (c *Client) DiscoveryClient() discovery.DiscoveryInterface {
 	return c.discoveryClient
 }
+
+func (c *Client) SetMaxPendingDuration(duration time.Duration) {
+	c.maxPendingDuration = duration
+}

pkg/k8s/namespace.go

@@ -24,10 +24,11 @@ func (c *Client) CreateNamespace(ctx context.Context, name string) error {
 		if !apierrs.IsAlreadyExists(err) {
 			return ErrCreatingNamespace.WithParams(name).Wrap(err)
 		}
-		c.logger.Debugf("Namespace %s already exists, continuing.\n", name)
+		c.logger.WithField("name", name).Debug("namespace already exists, continuing")
+		return nil
 	}
-	c.logger.Debugf("Namespace %s created.\n", name)
 
+	c.logger.WithField("name", name).Debug("namespace created")
 	return nil
 }
 
@@ -50,10 +51,10 @@ func (c *Client) NamespaceExists(ctx context.Context, name string) (bool, error)
 	}
 
 	if apierrs.IsNotFound(err) {
-		c.logger.Debugf("Namespace %s does not exist, err: %v", name, err)
+		c.logger.WithField("name", name).WithError(err).Debug("namespace does not exist")
 		return false, nil
 	}
 
-	c.logger.Errorf("Error getting namespace %s, err: %v", name, err)
+	c.logger.WithField("name", name).WithError(err).Error("getting namespace")
 	return false, err
 }

pkg/k8s/networkpolicy.go

@@ -97,7 +97,7 @@ func (c *Client) GetNetworkPolicy(ctx context.Context, name string) (*v1.Network
 func (c *Client) NetworkPolicyExists(ctx context.Context, name string) bool {
 	_, err := c.GetNetworkPolicy(ctx, name)
 	if err != nil {
-		c.logger.Debug("NetworkPolicy does not exist, err: ", err)
+		c.logger.WithField("name", name).WithError(err).Debug("getting networkPolicy")
 		return false
 	}
 

pkg/k8s/pod.go

@@ -9,6 +9,7 @@ import (
 	"strings"
 	"time"
 
+	"github.com/sirupsen/logrus"
 	v1 "k8s.io/api/core/v1"
 	apierrs "k8s.io/apimachinery/pkg/api/errors"
 	"k8s.io/apimachinery/pkg/api/resource"
@@ -105,7 +106,7 @@ func (c *Client) NewFile(source, dest string) *File {
 }
 
 func (c *Client) ReplacePodWithGracePeriod(ctx context.Context, podConfig PodConfig, gracePeriod *int64) (*v1.Pod, error) {
-	c.logger.Debugf("Replacing pod %s", podConfig.Name)
+	c.logger.WithField("name", podConfig.Name).Debug("replacing pod")
 
 	if err := c.DeletePodWithGracePeriod(ctx, podConfig.Name, gracePeriod); err != nil {
 		return nil, ErrDeletingPod.Wrap(err)
@@ -127,13 +128,13 @@ func (c *Client) waitForPodDeletion(ctx context.Context, name string) error {
 	for {
 		select {
 		case <-ctx.Done():
-			c.logger.Errorf("Context cancelled while waiting for pod %s to delete", name)
+			c.logger.WithField("name", name).Error("context cancelled while waiting for pod to delete")
 			return ctx.Err()
 		case <-time.After(retryInterval):
 			_, err := c.getPod(ctx, name)
 			if err != nil {
 				if apierrs.IsNotFound(err) {
-					c.logger.Debugf("Pod %s successfully deleted", name)
+					c.logger.WithField("name", name).Debug("pod successfully deleted")
 					return nil
 				}
 				return ErrWaitingForPodDeletion.WithParams(name).Wrap(err)
@@ -309,8 +310,11 @@ func (c *Client) PortForwardPod(
 	if stderr.Len() > 0 {
 		return ErrPortForwarding.WithParams(stderr.String())
 	}
-	c.logger.Debugf("Port forwarding from %d to %d", localPort, remotePort)
-	c.logger.Debugf("Port forwarding stdout: %v", stdout)
+	c.logger.WithFields(logrus.Fields{
+		"local_port":  localPort,
+		"remote_port": remotePort,
+		"stdout":      stdout.String(),
+	}).Debug("port forwarding")
 
 	// Start the port forwarding
 	go func() {
@@ -325,7 +329,10 @@ func (c *Client) PortForwardPod(
 	select {
 	case <-readyChan:
 		// Ready to forward
-		c.logger.Debugf("Port forwarding ready from %d to %d", localPort, remotePort)
+		c.logger.WithFields(logrus.Fields{
+			"local_port":  localPort,
+			"remote_port": remotePort,
+		}).Debug("port forwarding ready")
 	case err := <-errChan:
 		// if there's an error, return it
 		return ErrForwardingPorts.Wrap(err)
@@ -491,7 +498,7 @@ func (c *Client) buildInitContainerCommand(volumes []*Volume, files []*File) []s
 	fullCommand := strings.Join(cmds, "")
 	commands = append(commands, fullCommand)
 
-	c.logger.Debugf("Init container command: %s", fullCommand)
+	c.logger.WithField("command", fullCommand).Debug("init container command")
 	return commands
 }
 
@@ -582,6 +589,9 @@ func (c *Client) preparePod(spec PodConfig, init bool) *v1.Pod {
 		Spec: c.preparePodSpec(spec, init),
 	}
 
-	c.logger.Debugf("Prepared pod %s in namespace %s", spec.Name, spec.Namespace)
+	c.logger.WithFields(logrus.Fields{
+		"name":      spec.Name,
+		"namespace": spec.Namespace,
+	}).Debug("prepared pod")
 	return pod
 }