ok

Author: alexec

api/v1alpha1/source_statuses.go

@@ -10,7 +10,7 @@ import (
 
 type SourceStatuses map[string]SourceStatus // key is replica
 
-func (in SourceStatuses) Set(name string, replica int, msg string, rate resource.Quantity) {
+func (in SourceStatuses) IncrTotal(name string, replica int, msg string, rate resource.Quantity) {
 	x := in[name]
 	x.LastMessage = &Message{Data: trunc(msg), Time: metav1.Now()}
 	if x.Metrics == nil {
@@ -30,7 +30,7 @@ func (in SourceStatuses) Get(name string) SourceStatus {
 	return SourceStatus{}
 }
 
-func (in SourceStatuses) IncErrors(name string, replica int, err error) {
+func (in SourceStatuses) IncrErrors(name string, replica int, err error) {
 	x := in[name]
 	msg := err.Error()
 	x.LastError = &Error{Message: trunc(msg), Time: metav1.Now()}

api/v1alpha1/source_statuses_test.go

@@ -13,7 +13,7 @@ import (
 func TestSourceStatuses_Set(t *testing.T) {
 	ss := SourceStatuses{}
 
-	ss.Set("bar", 1, strings.Repeat("x", 33), resource.MustParse("1"))
+	ss.IncrTotal("bar", 1, strings.Repeat("x", 33), resource.MustParse("1"))
 
 	if assert.Len(t, ss, 1) {
 		s := ss["bar"]
@@ -26,7 +26,7 @@ func TestSourceStatuses_Set(t *testing.T) {
 		}
 	}
 
-	ss.Set("bar", 1, "bar", resource.MustParse("1"))
+	ss.IncrTotal("bar", 1, "bar", resource.MustParse("1"))
 
 	if assert.Len(t, ss, 1) {
 		s := ss["bar"]
@@ -39,7 +39,7 @@ func TestSourceStatuses_Set(t *testing.T) {
 		}
 	}
 
-	ss.Set("bar", 0, "foo", resource.MustParse("1"))
+	ss.IncrTotal("bar", 0, "foo", resource.MustParse("1"))
 
 	if assert.Len(t, ss, 1) {
 		s := ss["bar"]
@@ -52,7 +52,7 @@ func TestSourceStatuses_Set(t *testing.T) {
 		}
 	}
 
-	ss.Set("baz", 0, "foo", resource.MustParse("1"))
+	ss.IncrTotal("baz", 0, "foo", resource.MustParse("1"))
 
 	if assert.Len(t, ss, 2) {
 		s := ss["baz"]
@@ -68,13 +68,13 @@ func TestSourceStatuses_Set(t *testing.T) {
 func TestSourceStatuses_IncErrors(t *testing.T) {
 	ss := SourceStatuses{}
 	err := errors.New(strings.Repeat("x", 33))
-	ss.IncErrors("foo", 0, err)
+	ss.IncrErrors("foo", 0, err)
 	assert.Equal(t, uint64(1), ss["foo"].Metrics["0"].Errors)
 	assert.NotEmpty(t, ss["foo"].LastError.Message)
 	assert.NotEmpty(t, ss["foo"].LastError.Time)
-	ss.IncErrors("foo", 0, err)
+	ss.IncrErrors("foo", 0, err)
 	assert.Equal(t, uint64(2), ss["foo"].Metrics["0"].Errors)
-	ss.IncErrors("bar", 0, err)
+	ss.IncrErrors("bar", 0, err)
 	assert.Equal(t, uint64(1), ss["bar"].Metrics["0"].Errors)
 }
 

runner/sidecar/sidecar.go

@@ -339,9 +339,6 @@ func connectSources(ctx context.Context, toMain func([]byte) error) error {
 		retryPolicy := source.RetryPolicy
 		f := func(msg []byte) error {
 			rateCounter.Incr(1)
-			withLock(func() {
-				step.Status.SourceStatuses.Set(sourceName, replica, printable(msg), rateToResourceQuantity(rateCounter))
-			})
 			err := wait.ExponentialBackoff(wait.Backoff{
 				Duration: 100 * time.Millisecond,
 				Factor:   1.2,
@@ -360,14 +357,18 @@ func connectSources(ctx context.Context, toMain func([]byte) error) error {
 						default:
 							return false, nil
 						}
+					} else {
+						return true, nil
 					}
-					return true, nil
 				}
 			})
 			if err != nil {
-				withLock(func() { step.Status.SourceStatuses.IncErrors(sourceName, replica, err) })
+				withLock(func() { step.Status.SourceStatuses.IncrTotal(sourceName, replica, printable(msg), rateToResourceQuantity(rateCounter)) })
+			} else {
+				withLock(func() { step.Status.SourceStatuses.IncrErrors(sourceName, replica, err) })
 			}
 			return err
+
 		}
 		if x := source.Cron; x != nil {
 			_, err := crn.AddFunc(x.Schedule, func() {
@@ -765,10 +766,10 @@ func connectSink() (func([]byte) error, error) {
 			counter := rateCounters[sinkName]
 			counter.Incr(1)
 			withLock(func() {
-				step.Status.SinkStatues.Set(sinkName, replica, printable(msg), rateToResourceQuantity(counter))
+				step.Status.SinkStatues.IncrTotal(sinkName, replica, printable(msg), rateToResourceQuantity(counter))
 			})
 			if err := f(msg); err != nil {
-				withLock(func() { step.Status.SinkStatues.IncErrors(sinkName, replica, err) })
+				withLock(func() { step.Status.SinkStatues.IncrErrors(sinkName, replica, err) })
 				return err
 			}
 		}