Merge branch 'add-sink-metric' of https://github.com/odpf/meteor into…

… add-sink-metric

agent/agent.go

@@ -236,15 +236,13 @@ func (r *Agent) setupProcessor(ctx context.Context, pr recipe.PluginRecipe, str
 
 func (r *Agent) setupSink(ctx context.Context, sr recipe.PluginRecipe, stream *stream, recipe recipe.Recipe) (err error) {
 	var sink plugins.Syncer
-	success := true
-	pluginType := "sink"
+
 	if sink, err = r.sinkFactory.Get(sr.Name); err != nil {
 		return errors.Wrapf(err, "could not find sink \"%s\"", sr.Name)
 	}
 	if err = sink.Init(ctx, sr.Config); err != nil {
 		return errors.Wrapf(err, "could not initiate sink \"%s\"", sr.Name)
 	}
-
 	retryNotification := func(e error, d time.Duration) {
 		r.logger.Info(
 			fmt.Sprintf("retrying sink in %d", d),
@@ -256,19 +254,22 @@ func (r *Agent) setupSink(ctx context.Context, sr recipe.PluginRecipe, stream *s
 			err := sink.Sink(ctx, records)
 			return err
 		}, retryNotification)
-		success = err == nil
-		r.monitor.RecordPlugin(recipe.Name, sr.Name, pluginType, success)
 
-		// error (after exhausted retries) will just be skipped and logged
+		var success bool
 		if err != nil {
+			// once it reaches here, it means that the retry has been exhausted and still got error
+			success = false
 			r.logger.Error("error running sink", "sink", sr.Name, "error", err.Error())
-			if !r.stopOnSinkError {
-				err = nil
-			}
-			return err
+		} else {
+			success = true
+			r.logger.Info("Successfully published record", "sink", sr.Name, "recipe", recipe.Name)
 		}
-		r.logger.Info("Successfully published record", "sink", sr.Name, "recipe", recipe.Name)
 
+		r.monitor.RecordPlugin(recipe.Name, sr.Name, "sink", success)
+
+		if !r.stopOnSinkError {
+			err = nil
+		}
 		// TODO: create a new error to signal stopping stream.
 		// returning nil so stream wont stop.
 		return err