Merge pull request #193 from libp2p/deferred-qlog-compression

compress qlogs when the QUIC connection is closed

p2p/transport/quic/tracer.go

@@ -37,10 +37,14 @@ func initQlogger(qlogDir string) logging.Tracer {
 	})
 }
 
+// The qlogger logs qlog events to a temporary file: .<name>.qlog.swp.
+// When it is closed, it compresses the temporary file and saves it as <name>.qlog.zst.
+// It is not possible to compress on the fly, as compression algorithms keep a lot of internal state,
+// which can easily exhaust the host system's memory when running a few hundred QUIC connections in parallel.
 type qlogger struct {
-	f        *os.File // QLOGDIR/.log_xxx.qlog.gz.swp
-	filename string   // QLOGDIR/log_xxx.qlog.gz
-	io.WriteCloser
+	f             *os.File // QLOGDIR/.log_xxx.qlog.swp
+	filename      string   // QLOGDIR/log_xxx.qlog.zst
+	*bufio.Writer          // buffering the f
 }
 
 func newQlogger(qlogDir string, role logging.Perspective, connID []byte) io.WriteCloser {
@@ -50,50 +54,43 @@ func newQlogger(qlogDir string, role logging.Perspective, connID []byte) io.Writ
 		r = "client"
 	}
 	finalFilename := fmt.Sprintf("%s%clog_%s_%s_%x.qlog.zst", qlogDir, os.PathSeparator, t, r, connID)
-	filename := fmt.Sprintf("%s%c.log_%s_%s_%x.qlog.zst.swp", qlogDir, os.PathSeparator, t, r, connID)
+	filename := fmt.Sprintf("%s%c.log_%s_%s_%x.qlog.swp", qlogDir, os.PathSeparator, t, r, connID)
 	f, err := os.Create(filename)
 	if err != nil {
 		log.Errorf("unable to create qlog file %s: %s", filename, err)
 		return nil
 	}
-	gz, err := zstd.NewWriter(f, zstd.WithEncoderLevel(zstd.SpeedFastest))
-	if err != nil {
-		log.Errorf("failed to initialize zstd: %s", err)
-		return nil
-	}
 	return &qlogger{
-		f:           f,
-		filename:    finalFilename,
-		WriteCloser: newBufferedWriteCloser(bufio.NewWriter(gz), gz),
+		f:        f,
+		filename: finalFilename,
+		Writer:   bufio.NewWriter(f),
 	}
 }
 
 func (l *qlogger) Close() error {
-	if err := l.WriteCloser.Close(); err != nil {
+	defer os.Remove(l.f.Name())
+	defer l.f.Close()
+	if err := l.Writer.Flush(); err != nil {
 		return err
 	}
-	path := l.f.Name()
-	if err := l.f.Close(); err != nil {
+	if _, err := l.f.Seek(0, io.SeekStart); err != nil { // set the read position to the beginning of the file
 		return err
 	}
-	return os.Rename(path, l.filename)
-}
-
-type bufferedWriteCloser struct {
-	*bufio.Writer
-	io.Closer
-}
-
-func newBufferedWriteCloser(writer *bufio.Writer, closer io.Closer) io.WriteCloser {
-	return &bufferedWriteCloser{
-		Writer: writer,
-		Closer: closer,
+	f, err := os.Create(l.filename)
+	if err != nil {
+		return err
 	}
-}
-
-func (h bufferedWriteCloser) Close() error {
-	if err := h.Writer.Flush(); err != nil {
+	defer f.Close()
+	buf := bufio.NewWriter(f)
+	c, err := zstd.NewWriter(buf, zstd.WithEncoderLevel(zstd.SpeedFastest))
+	if err != nil {
+		return err
+	}
+	if _, err := io.Copy(c, l.f); err != nil {
+		return err
+	}
+	if err := c.Close(); err != nil {
 		return err
 	}
-	return h.Closer.Close()
+	return buf.Flush()
 }

p2p/transport/quic/tracer_test.go

@@ -14,10 +14,6 @@ import (
 	. "github.com/onsi/gomega"
 )
 
-type nopCloser struct{}
-
-func (nopCloser) Close() error { return nil }
-
 var _ = Describe("qlogger", func() {
 	var qlogDir string
 
@@ -44,7 +40,7 @@ var _ = Describe("qlogger", func() {
 		logger := newQlogger(qlogDir, logging.PerspectiveServer, []byte{0xde, 0xad, 0xbe, 0xef})
 		file := getFile()
 		Expect(string(file.Name()[0])).To(Equal("."))
-		Expect(file.Name()).To(HaveSuffix(".qlog.zst.swp"))
+		Expect(file.Name()).To(HaveSuffix(".qlog.swp"))
 		// close the logger. This should move the file.
 		Expect(logger.Close()).To(Succeed())
 		file = getFile()
@@ -77,9 +73,9 @@ var _ = Describe("qlogger", func() {
 		compressed, err := ioutil.ReadFile(qlogDir + "/" + getFile().Name())
 		Expect(err).ToNot(HaveOccurred())
 		Expect(compressed).ToNot(Equal("foobar"))
-		gz, err := zstd.NewReader(bytes.NewReader(compressed))
+		c, err := zstd.NewReader(bytes.NewReader(compressed))
 		Expect(err).ToNot(HaveOccurred())
-		data, err := ioutil.ReadAll(gz)
+		data, err := ioutil.ReadAll(c)
 		Expect(err).ToNot(HaveOccurred())
 		Expect(data).To(Equal([]byte("foobar")))
 	})