Merge pull request #3890 from gyuho/pipeline_full_message

rafthttp: more detailed logs when filled-up buffer

rafthttp/peer.go

@@ -171,10 +171,9 @@ func startPeer(streamRt, pipelineRt http.RoundTripper, urls types.URLs, local, t
 						p.r.ReportSnapshot(m.To, raft.SnapshotFailure)
 					}
 					if status.isActive() {
-						plog.MergeWarningf("dropped %s to %s since %s's sending buffer is full", m.Type, p.id, name)
-					} else {
-						plog.Debugf("dropped %s to %s since %s's sending buffer is full", m.Type, p.id, name)
+						plog.MergeWarningf("dropped internal raft message to %s since %s's sending buffer is full (bad/overloaded network)", p.id, name)
 					}
+					plog.Debugf("dropped %s to %s since %s's sending buffer is full", m.Type, p.id, name)
 				}
 			case mm := <-p.recvc:
 				if err := r.Process(context.TODO(), mm); err != nil {

rafthttp/remote.go

@@ -42,10 +42,9 @@ func (g *remote) send(m raftpb.Message) {
 	case g.pipeline.msgc <- m:
 	default:
 		if g.status.isActive() {
-			plog.MergeWarningf("dropped %s to %s since sending buffer is full", m.Type, g.id)
-		} else {
-			plog.Debugf("dropped %s to %s since sending buffer is full", m.Type, g.id)
+			plog.MergeWarningf("dropped internal raft message to %s since sending buffer is full (bad/overloaded network)", g.id)
 		}
+		plog.Debugf("dropped %s to %s since sending buffer is full", m.Type, g.id)
 	}
 }
 

rafthttp/stream.go

@@ -332,10 +332,9 @@ func (cr *streamReader) decodeLoop(rc io.ReadCloser, t streamType) error {
 		case recvc <- m:
 		default:
 			if cr.status.isActive() {
-				plog.MergeWarningf("dropped %s from %s since receiving buffer is full", m.Type, types.ID(m.From))
-			} else {
-				plog.Debugf("dropped %s from %s since receiving buffer is full", m.Type, types.ID(m.From))
+				plog.MergeWarningf("dropped internal raft message from %s since receiving buffer is full (overloaded network)", types.ID(m.From))
 			}
+			plog.Debugf("dropped %s from %s since receiving buffer is full", m.Type, types.ID(m.From))
 		}
 	}
 }