Compress backups before transmission between nodes

cluster/client.go

@@ -1,6 +1,8 @@
 package cluster
 
 import (
+	"bytes"
+	"compress/gzip"
 	"encoding/binary"
 	"errors"
 	"fmt"
@@ -360,6 +362,13 @@ func (c *Client) Backup(br *command.BackupRequest, nodeAddr string, creds *Crede
 		return err
 	}
 
+	// Decompress....
+	p, err = gzUncompress(p)
+	if err != nil {
+		handleConnError(conn)
+		return err
+	}
+
 	resp := &CommandBackupResponse{}
 	err = proto.Unmarshal(p, resp)
 	if err != nil {
@@ -447,3 +456,20 @@ func handleConnError(conn net.Conn) {
 		pc.MarkUnusable()
 	}
 }
+
+func gzUncompress(b []byte) ([]byte, error) {
+	gz, err := gzip.NewReader(bytes.NewReader(b))
+	if err != nil {
+		return nil, fmt.Errorf("unmarshal gzip NewReader: %s", err)
+	}
+
+	ub, err := io.ReadAll(gz)
+	if err != nil {
+		return nil, fmt.Errorf("unmarshal gzip ReadAll: %s", err)
+	}
+
+	if err := gz.Close(); err != nil {
+		return nil, fmt.Errorf("unmarshal gzip Close: %s", err)
+	}
+	return ub, nil
+}

cluster/service.go

@@ -2,6 +2,7 @@ package cluster
 
 import (
 	"bytes"
+	"compress/gzip"
 	"encoding/binary"
 	"expvar"
 	"fmt"
@@ -317,7 +318,6 @@ func (s *Service) handleConn(conn net.Conn) {
 				} else {
 					resp.Data = buf.Bytes()
 				}
-
 			}
 			p, err = proto.Marshal(resp)
 			if err != nil {
@@ -325,6 +325,13 @@ func (s *Service) handleConn(conn net.Conn) {
 				return
 			}
 
+			// Compress the backup.
+			p, err = gzCompress(p)
+			if err != nil {
+				conn.Close()
+				return
+			}
+
 			// Write length of Protobuf first, then write the actual Protobuf.
 			b = make([]byte, 4)
 			binary.LittleEndian.PutUint32(b[0:], uint32(len(p)))
@@ -333,3 +340,20 @@ func (s *Service) handleConn(conn net.Conn) {
 		}
 	}
 }
+
+// gzCompress compresses the given byte slice.
+func gzCompress(b []byte) ([]byte, error) {
+	var buf bytes.Buffer
+	gzw, err := gzip.NewWriterLevel(&buf, gzip.BestCompression)
+	if err != nil {
+		return nil, fmt.Errorf("gzip new writer: %s", err)
+	}
+
+	if _, err := gzw.Write(b); err != nil {
+		return nil, fmt.Errorf("gzip Write: %s", err)
+	}
+	if err := gzw.Close(); err != nil {
+		return nil, fmt.Errorf("gzip Close: %s", err)
+	}
+	return buf.Bytes(), nil
+}