chore: refine error messages (#467)

* chore: refine error messages Signed-off-by: Derek Wang <[email protected]> * . Signed-off-by: Derek Wang <[email protected]> * . Signed-off-by: Derek Wang <[email protected]>
argoproj-labs · Oct 21, 2021 · 8a793ec · 8a793ec
1 parent 9014b30
commit 8a793ec
Show file tree

Hide file tree

Showing 3 changed files with 14 additions and 4 deletions.
diff --git a/runner/sidecar/sink/stan/stan.go b/runner/sidecar/sink/stan/stan.go
@@ -56,7 +56,7 @@ func New(ctx context.Context, secretInterface corev1.SecretInterface, namespace,
 					clientID := genClientID()
 					conn, err = stan.ConnectSTAN(ctx, secretInterface, x, clientID)
 					if err != nil {
-						logger.Error(err, "failed to reconnect", "sink", sinkName, "clientID", clientID)
+						logger.Info("failed to reconnect, will try again soon", "sink", sinkName, "clientID", clientID, "error", err)
 						continue
 					}
 					logger.Info("reconnected to stan server.", "sink", sinkName, "clientID", clientID)

diff --git a/runner/sidecar/source/jetstream/jetstream.go b/runner/sidecar/source/jetstream/jetstream.go
@@ -2,6 +2,7 @@ package js
 
 import (
 	"context"
+	"errors"
 	"fmt"
 
 	dfv1 "github.com/argoproj-labs/argo-dataflow/api/v1alpha1"
@@ -43,7 +44,11 @@ func New(ctx context.Context, secretInterface corev1.SecretInterface, cluster, n
 			); err != nil {
 				logger.Error(err, "failed to process message")
 			} else if err := msg.Ack(); err != nil {
-				logger.Error(err, "failed to ack message", "source", sourceName)
+				if errors.Is(err, nats.ErrBadSubscription) {
+					logger.Info("Jet Stream subscription might have been closed", "source", sourceName, "error", err)
+				} else {
+					logger.Error(err, "failed to ack message", "source", sourceName)
+				}
 			}
 		}
 	}, nats.ManualAck(), nats.Durable(durableName), nats.DeliverNew())

diff --git a/runner/sidecar/source/stan/stan.go b/runner/sidecar/source/stan/stan.go
@@ -3,6 +3,7 @@ package stan
 import (
 	"context"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"math/rand"
 	"net/http"
@@ -62,7 +63,11 @@ func New(ctx context.Context, secretInterface corev1.SecretInterface, cluster, n
 			); err != nil {
 				logger.Error(err, "failed to process message")
 			} else if err := msg.Ack(); err != nil {
-				logger.Error(err, "failed to ack message", "source", sourceName)
+				if errors.Is(err, stan.ErrBadSubscription) {
+					logger.Info("failed to ack a message, stan subscription might have been closed", "source", sourceName, "error", err)
+				} else {
+					logger.Error(err, "failed to ack a message", "source", sourceName)
+				}
 			}
 		}, stan.DurableName(queueName),
 			stan.SetManualAckMode(),
@@ -96,7 +101,7 @@ func New(ctx context.Context, secretInterface corev1.SecretInterface, cluster, n
 					clientID := genClientID()
 					conn, err = sharedstan.ConnectSTAN(ctx, secretInterface, x, clientID)
 					if err != nil {
-						logger.Error(err, "failed to reconnect", "source", sourceName, "clientID", clientID)
+						logger.Info("failed to reconnect, will try again soon", "source", sourceName, "clientID", clientID, "error", err)
 						continue
 					}
 					logger.Info("reconnected to stan server.", "source", sourceName, "clientID", clientID)