fix(neuron): avoid flooding logs when disconnect (#3393)

Signed-off-by: Jiyong Huang <[email protected]>

internal/io/neuron/source.go

@@ -94,16 +94,21 @@ func (s *source) Subscribe(ctx api.StreamContext, ingest api.BytesIngest, ingest
 	ctx.GetLogger().Infof("neuron source receiving loop started")
 	go func() {
 		err := infra.SafeRun(func() error {
+			connected := true
 			for {
 				// no receiving deadline, will wait until the socket closed
 				if msg, err := s.cli.Recv(); err == nil {
+					connected = true
 					ctx.GetLogger().Debugf("nng received message %s", string(msg))
 					rawData, meta := extractTraceMeta(ctx, msg)
 					ingest(ctx, rawData, meta, timex.GetNow())
 				} else if err == mangos.ErrClosed {
-					ctx.GetLogger().Infof("neuron connection closed, retry after 1 second")
-					ingestErr(ctx, errors.New("neuron connection closed"))
-					time.Sleep(1 * time.Second)
+					if connected {
+						ctx.GetLogger().Infof("neuron connection closed, retry after 1 second")
+						ingestErr(ctx, errors.New("neuron connection closed"))
+						time.Sleep(1 * time.Second)
+						connected = false
+					}
 					continue
 				}
 			}