Repository: numaproj/numaflow
Pull request #650 (2023-03-30 21:37:39)

feat: integrate serde WAL
continued [TODO: integrate with WAL segment](https://github.com/numaproj/numaflow/pull/565)

closes https://github.com/numaproj/numaflow/issues/488


# go profiling
WAL.Write 
CPU comparison
gob 20.32% vs serde 11.54%
For the heap I don't see WAL anywhere. I added the go profiling heap graph below.

## CPU comparison
### gob
<img width="1792" alt="Screenshot 2023-03-31 at 11 57 04 AM" src="https://user-images.githubusercontent.com/19543684/229206692-b7714572-4e0e-43b8-9aef-41a52c18c28d.png">
<img width="1792" alt="Screenshot 2023-03-31 at 11 57 17 AM" src="https://user-images.githubusercontent.com/19543684/229206701-f2325cba-bdbe-46ec-b1a1-c198aa8de28a.png">
<img width="1792" alt="Screenshot 2023-03-31 at 11 57 23 AM" src="https://user-images.githubusercontent.com/19543684/229206706-5d3a5541-5a67-4049-ad93-2461acc78290.png">
<img width="1792" alt="Screenshot 2023-03-31 at 11 57 27 AM" src="https://user-images.githubusercontent.com/19543684/229206710-3d10f7e1-2035-4612-a1bd-b8687c5c7f6b.png">

### serde
<img width="1792" alt="Screenshot 2023-03-31 at 11 43 18 AM" src="https://user-images.githubusercontent.com/19543684/229206736-646221ca-3dda-4eed-9054-47fccb173d70.png">

## MEM comparison
### gob
<img width="1792" alt="Screenshot 2023-03-31 at 11 55 43 AM" src="https://user-images.githubusercontent.com/19543684/229206824-2085e161-151d-48be-b978-9198637961b9.png">
<img width="1792" alt="Screenshot 2023-03-31 at 11 55 56 AM" src="https://user-images.githubusercontent.com/19543684/229206829-49bb7028-b839-462f-bf4c-4a119b6fa362.png">

### serde
<img width="1792" alt="Screenshot 2023-03-31 at 11 45 46 AM" src="https://user-images.githubusercontent.com/19543684/229206867-9a16b0ea-b748-4017-b02c-6087c1d973fb.png">







Patch:
diff --git a/.github/workflows/ci.yaml b/.github/workflows/ci.yaml
index 60fbb5d7..61b17aa5 100644
--- a/.github/workflows/ci.yaml
+++ b/.github/workflows/ci.yaml
@@ -40,9 +40,9 @@ jobs:
           path: ~/.cache/go-build
           key: ${{ runner.os }}-go-build-v1-${{ hashFiles('**/go.mod') }}
       - name: Setup Golang
-        uses: actions/setup-go@v3.1.0
+        uses: actions/setup-go@v4.0.0
         with:
-          go-version: "1.19"
+          go-version: '1.19'
       - name: Add bins to PATH
         run: |
           echo /home/runner/go/bin >> $GITHUB_PATH
@@ -94,9 +94,9 @@ jobs:
           --health-retries 5
     steps:
       - name: Set up Go 1.x
-        uses: actions/setup-go@v3.1.0
+        uses: actions/setup-go@v4.0.0
         with:
-          go-version: "1.19"
+          go-version: '1.19'
         id: go
       - name: Check out code into the Go module directory
         uses: actions/checkout@v3
@@ -119,9 +119,9 @@ jobs:
     steps:
       - uses: actions/checkout@v3
       - name: Setup Golang
-        uses: actions/setup-go@v3.1.0
+        uses: actions/setup-go@v4.0.0
         with:
-          go-version: "1.19"
+          go-version: '1.19'
       - name: Restore Go build cache
         uses: actions/cache@v3
         with:
@@ -139,7 +139,7 @@ jobs:
       max-parallel: 7
       matrix:
         driver: [jetstream]
-        case: [e2e-suite-1, e2e-suite-2, kafka-e2e, http-e2e, nats-e2e, redis-source-e2e, sdks-e2e, reduce-e2e]
+        case: [e2e-suite-1, e2e-suite-2, kafka-e2e, http-e2e, nats-e2e, redis-streams-e2e, sdks-e2e, reduce-e2e]
     steps:
       - name: Checkout code
         uses: actions/checkout@v3
@@ -149,9 +149,9 @@ jobs:
           path: ~/.cache/go-build
           key: ${{ runner.os }}-go-build-v2-${{ hashFiles('**/go.mod') }}
       - name: Setup Golang
-        uses: actions/setup-go@v3.1.0
+        uses: actions/setup-go@v4.0.0
         with:
-          go-version: "1.19"
+          go-version: '1.19'
       - name: Add bins to PATH
         run: |
           echo /home/runner/go/bin >> $GITHUB_PATH
diff --git a/.github/workflows/nightly-build.yml b/.github/workflows/nightly-build.yml
index 4ee6f61d..2a9b7b89 100644
--- a/.github/workflows/nightly-build.yml
+++ b/.github/workflows/nightly-build.yml
@@ -24,9 +24,9 @@ jobs:
         uses: actions/checkout@v3
 
       - name: Setup Go
-        uses: actions/setup-go@v3.1.0
+        uses: actions/setup-go@v4.0.0
         with:
-          go-version: 1.19
+          go-version: '1.19'
 
       - name: Build binaries
         run: |
diff --git a/.github/workflows/release.yml b/.github/workflows/release.yml
index 68be289a..e4b77d49 100644
--- a/.github/workflows/release.yml
+++ b/.github/workflows/release.yml
@@ -22,9 +22,9 @@ jobs:
         uses: actions/checkout@v3
 
       - name: Setup Go
-        uses: actions/setup-go@v3.1.0
+        uses: actions/setup-go@v4.0.0
         with:
-          go-version: 1.19
+          go-version: '1.19'
 
       - name: Build binaries
         run: |
@@ -99,9 +99,9 @@ jobs:
           else
             echo "VERSION=${GITHUB_REF##*/}" >> $GITHUB_ENV
           fi
-      - uses: actions/setup-go@v3.1.0
+      - uses: actions/setup-go@v4.0.0
         with:
-          go-version: 1.19
+          go-version: '1.19'
       - uses: actions/checkout@v3
       - run: go install sigs.k8s.io/bom/cmd/bom@v0.2.0
       - run: go install github.com/spdx/spdx-sbom-generator/cmd/generator@v0.0.13
diff --git a/Makefile b/Makefile
index 8f15768b..51f4915f 100644
--- a/Makefile
+++ b/Makefile
@@ -110,7 +110,7 @@ test-e2e-suite-2:
 test-kafka-e2e:
 test-http-e2e:
 test-nats-e2e:
-test-redis-source-e2e:
+test-redis-streams-e2e:
 test-sdks-e2e:
 test-reduce-e2e:
 test-%: 
diff --git a/pkg/reduce/pbq/store/wal/bootstrap.go b/pkg/reduce/pbq/store/wal/bootstrap.go
index 62d4ce07..3dd64b25 100644
--- a/pkg/reduce/pbq/store/wal/bootstrap.go
+++ b/pkg/reduce/pbq/store/wal/bootstrap.go
@@ -17,9 +17,7 @@ limitations under the License.
 package wal
 
 import (
-	"bytes"
 	"encoding/binary"
-	"encoding/gob"
 	"fmt"
 	"io"
 	"os"
@@ -133,14 +131,14 @@ func (w *WAL) Read(size int64) ([]*isb.ReadMessage, bool, error) {
 	return messages, false, nil
 }
 
-// decodeReadMessage decodes the header which is encoded by encodeReadMessage.
+// decodeReadMessage decodes the WALMessage which is encoded by encodeWALMessage.
 func decodeReadMessage(buf io.Reader) (*isb.ReadMessage, int64, error) {
-	entryHeader, err := decodeReadMessageHeader(buf)
+	entryHeader, err := decodeWALMessageHeader(buf)
 	if err != nil {
 		return nil, 0, err
 	}
 
-	entryBody, err := decodeReadMessageBody(buf, entryHeader)
+	entryBody, err := decodeWALBody(buf, entryHeader)
 	if err != nil {
 		return nil, 0, err
 	}
@@ -153,8 +151,8 @@ func decodeReadMessage(buf io.Reader) (*isb.ReadMessage, int64, error) {
 	}, size, nil
 }
 
-// decodeReadMessageHeader decodes the header which is encoded by encodeReadMessageHeader.
-func decodeReadMessageHeader(buf io.Reader) (*readMessageHeaderPreamble, error) {
+// decodeWALMessageHeader decodes the WALMessage header which is encoded by encodeWALMessageHeader.
+func decodeWALMessageHeader(buf io.Reader) (*readMessageHeaderPreamble, error) {
 	// read the fixed vals
 	var entryHeader = new(readMessageHeaderPreamble)
 	err := binary.Read(buf, binary.LittleEndian, entryHeader)
@@ -164,9 +162,9 @@ func decodeReadMessageHeader(buf io.Reader) (*readMessageHeaderPreamble, error)
 	return entryHeader, nil
 }
 
-// decodeReadMessageBody decodes the header which is encoded by encodeReadMessageBody.
+// decodeWALBody decodes the WALMessage body which is encoded by encodeWALMessageBody.
 // Returns errChecksumMismatch to indicate if corrupted entry is found.
-func decodeReadMessageBody(buf io.Reader, entryHeader *readMessageHeaderPreamble) (*isb.Message, error) {
+func decodeWALBody(buf io.Reader, entryHeader *readMessageHeaderPreamble) (*isb.Message, error) {
 	var err error
 
 	body := make([]byte, entryHeader.MessageLen)
@@ -185,8 +183,7 @@ func decodeReadMessageBody(buf io.Reader, entryHeader *readMessageHeaderPreamble
 	}
 
 	var message = new(isb.Message)
-	dec := gob.NewDecoder(bytes.NewReader(body))
-	err = dec.Decode(&message)
+	err = message.UnmarshalBinary(body)
 	if err != nil {
 		return nil, err
 	}
diff --git a/pkg/reduce/pbq/store/wal/segment.go b/pkg/reduce/pbq/store/wal/segment.go
index b05aa029..ef3fb80a 100644
--- a/pkg/reduce/pbq/store/wal/segment.go
+++ b/pkg/reduce/pbq/store/wal/segment.go
@@ -19,7 +19,6 @@ package wal
 import (
 	"bytes"
 	"encoding/binary"
-	"encoding/gob"
 	"fmt"
 	"hash/crc32"
 	"os"
@@ -153,26 +152,27 @@ func (w *WAL) encodeWALHeader(id *partition.ID) (buf *bytes.Buffer, err error) {
 	return buf, err
 }
 
-func (w *WAL) encodeReadMessage(message *isb.ReadMessage) (buf *bytes.Buffer, err error) {
+// encodeWALMessage builds the WAL message.
+func (w *WAL) encodeWALMessage(message *isb.ReadMessage) (buf *bytes.Buffer, err error) {
 	defer func() {
 		if err != nil {
 			walErrors.With(map[string]string{
 				metrics.LabelPipeline:           w.walStores.pipelineName,
 				metrics.LabelVertex:             w.walStores.vertexName,
 				metrics.LabelVertexReplicaIndex: strconv.Itoa(int(w.walStores.replicaIndex)),
-				labelErrorKind:                  "encodeReadMessage",
+				labelErrorKind:                  "encodeWALMessage",
 			}).Inc()
 		}
 	}()
 	buf = new(bytes.Buffer)
-	body, err := w.encodeReadMessageBody(message)
+	body, err := w.encodeWALMessageBody(message)
 	if err != nil {
 		return nil, err
 	}
-	checksum := calculateChecksum(body.Bytes())
+	checksum := calculateChecksum(body)
 
 	// Writes the message header
-	header, err := w.encodeReadMessageHeader(message, int64(body.Len()), checksum)
+	header, err := w.encodeWALMessageHeader(message, int64(len(body)), checksum)
 	if err != nil {
 		return nil, err
 	}
@@ -185,17 +185,19 @@ func (w *WAL) encodeReadMessage(message *isb.ReadMessage) (buf *bytes.Buffer, er
 	}
 
 	// Writes the message body
-	wrote, err = buf.Write(body.Bytes())
+	wrote, err = buf.Write(body)
 	if err != nil {
 		return nil, err
 	}
-	if wrote != body.Len() {
+	if wrote != len(body) {
 		return nil, fmt.Errorf("expected to write %d, but wrote only %d, %w", header.Len(), wrote, err)
 	}
 	return buf, nil
 }
 
-func (w *WAL) encodeReadMessageHeader(message *isb.ReadMessage, messageLen int64, checksum uint32) (*bytes.Buffer, error) {
+// encodeWALMessageHeader creates the header of the WAL message. The header is mainly for checksum, verifying the
+// correctness of the WALMessage.
+func (w *WAL) encodeWALMessageHeader(message *isb.ReadMessage, messageLen int64, checksum uint32) (*bytes.Buffer, error) {
 	watermark := message.Watermark.UnixMilli()
 
 	offset, err := message.ReadOffset.Sequence()
@@ -218,27 +220,27 @@ func (w *WAL) encodeReadMessageHeader(message *isb.ReadMessage, messageLen int64
 			metrics.LabelPipeline:           w.walStores.pipelineName,
 			metrics.LabelVertex:             w.walStores.vertexName,
 			metrics.LabelVertexReplicaIndex: strconv.Itoa(int(w.walStores.replicaIndex)),
-			labelErrorKind:                  "encodeReadMessageHeader",
+			labelErrorKind:                  "encodeWALMessageHeader",
 		}).Inc()
 		return nil, err
 	}
 	return buf, nil
 }
 
-func (w *WAL) encodeReadMessageBody(message *isb.ReadMessage) (*bytes.Buffer, error) {
-	m := new(bytes.Buffer)
-	enc := gob.NewEncoder(m)
-	err := enc.Encode(message.Message)
+// encodeWALMessageBody uses ReadMessage.Message field as the body of the WAL message, encodes the
+// ReadMessage.Message, and returns.
+func (w *WAL) encodeWALMessageBody(readMsg *isb.ReadMessage) ([]byte, error) {
+	msgBinary, err := readMsg.Message.MarshalBinary()
 	if err != nil {
 		walErrors.With(map[string]string{
 			metrics.LabelPipeline:           w.walStores.pipelineName,
 			metrics.LabelVertex:             w.walStores.vertexName,
 			metrics.LabelVertexReplicaIndex: strconv.Itoa(int(w.walStores.replicaIndex)),
-			labelErrorKind:                  "encodeReadMessageBody",
+			labelErrorKind:                  "encodeWALMessageBody",
 		}).Inc()
-		return nil, fmt.Errorf("encodeReadMessageBody encountered encode err: %w", err)
+		return nil, fmt.Errorf("encodeWALMessageBody encountered encode err: %w", err)
 	}
-	return m, nil
+	return msgBinary, nil
 }
 
 func calculateChecksum(data []byte) uint32 {
@@ -265,7 +267,7 @@ func (w *WAL) Write(message *isb.ReadMessage) (err error) {
 		}
 	}()
 	encodeStart := time.Now()
-	entry, err := w.encodeReadMessage(message)
+	entry, err := w.encodeWALMessage(message)
 	entryEncodeLatency.With(map[string]string{
 		metrics.LabelPipeline:           w.walStores.pipelineName,
 		metrics.LabelVertex:             w.walStores.vertexName,


Test patch:
diff --git a/pkg/reduce/pbq/store/wal/segment_test.go b/pkg/reduce/pbq/store/wal/segment_test.go
index 1d23c8b4..721d3aea 100644
--- a/pkg/reduce/pbq/store/wal/segment_test.go
+++ b/pkg/reduce/pbq/store/wal/segment_test.go
@@ -165,7 +165,7 @@ func Test_writeReadEntry(t *testing.T) {
 	actualOffset, err := actualMessage.ReadOffset.Sequence()
 	assert.NoError(t, err)
 	assert.Equalf(t, expectedOffset, actualOffset, "Read(%v)", message.ReadOffset)
-	assert.Equalf(t, message.Watermark, actualMessage.Watermark, "encodeReadMessage(%v)", message.Watermark)
+	assert.Equalf(t, message.Watermark, actualMessage.Watermark, "encodeWALMessage(%v)", message.Watermark)
 
 	// Start to write an entry again
 	err = newWal.Write(&message)
@@ -210,20 +210,20 @@ func Test_encodeDecodeEntry(t *testing.T) {
 			assert.NoError(t, err)
 			newWal := wal.(*WAL)
 
-			got, err := newWal.encodeReadMessage(tt.message)
-			if !tt.wantErr(t, err, fmt.Sprintf("encodeReadMessage(%v)", tt.message)) {
+			got, err := newWal.encodeWALMessage(tt.message)
+			if !tt.wantErr(t, err, fmt.Sprintf("encodeWALMessage(%v)", tt.message)) {
 				return
 			}
 
 			result, _, err := decodeReadMessage(bytes.NewReader(got.Bytes()))
 			assert.NoError(t, err)
-			assert.Equalf(t, tt.message.Message, result.Message, "encodeReadMessage(%v)", tt.message.Message)
+			assert.Equalf(t, tt.message.Message, result.Message, "encodeWALMessage(%v)", tt.message.Message)
 			expectedOffset, err := tt.message.ReadOffset.Sequence()
 			assert.NoError(t, err)
 			actualOffset, err := result.ReadOffset.Sequence()
 			assert.NoError(t, err)
-			assert.Equalf(t, expectedOffset, actualOffset, "encodeReadMessage(%v)", tt.message.ReadOffset)
-			assert.Equalf(t, tt.message.Watermark, result.Watermark, "encodeReadMessage(%v)", tt.message.Watermark)
+			assert.Equalf(t, expectedOffset, actualOffset, "encodeWALMessage(%v)", tt.message.ReadOffset)
+			assert.Equalf(t, tt.message.Watermark, result.Watermark, "encodeWALMessage(%v)", tt.message.Watermark)
 			err = newWal.Close()
 			assert.NoError(t, err)
 		})
@@ -259,7 +259,7 @@ func Test_batchSyncWithMaxBatchSize(t *testing.T) {
 	assert.NoError(t, err)
 	err = wal.Write(&message)
 	assert.NoError(t, err)
-	assert.Equal(t, int64(418), tempWAL.prevSyncedWOffset)
+	assert.Equal(t, int64(214), tempWAL.prevSyncedWOffset)
 
 	err = wal.Close()
 	assert.NoError(t, err)
@@ -286,7 +286,7 @@ func Test_batchSyncWithMaxBatchSize(t *testing.T) {
 	actualOffset, err := actualMessage.ReadOffset.Sequence()
 	assert.NoError(t, err)
 	assert.Equalf(t, expectedOffset, actualOffset, "Read(%v)", message.ReadOffset)
-	assert.Equalf(t, message.Watermark, actualMessage.Watermark, "encodeReadMessage(%v)", message.Watermark)
+	assert.Equalf(t, message.Watermark, actualMessage.Watermark, "encodeWALMessage(%v)", message.Watermark)
 
 	// Start to write an entry again
 	err = newWal.Write(&message)
@@ -316,7 +316,7 @@ func Test_batchSyncWithSyncDuration(t *testing.T) {
 	message := writeMessages[0]
 	storePrevSyncedTime := tempWAL.prevSyncedTime
 	err = wal.Write(&message)
-	assert.Equal(t, int64(228), tempWAL.prevSyncedWOffset)
+	assert.Equal(t, int64(126), tempWAL.prevSyncedWOffset)
 	assert.NotEqual(t, storePrevSyncedTime, tempWAL.prevSyncedTime)
 	assert.NoError(t, err)
 
@@ -352,7 +352,7 @@ func Test_batchSyncWithSyncDuration(t *testing.T) {
 	actualOffset, err := actualMessage.ReadOffset.Sequence()
 	assert.NoError(t, err)
 	assert.Equalf(t, expectedOffset, actualOffset, "Read(%v)", message.ReadOffset)
-	assert.Equalf(t, message.Watermark, actualMessage.Watermark, "encodeReadMessage(%v)", message.Watermark)
+	assert.Equalf(t, message.Watermark, actualMessage.Watermark, "encodeWALMessage(%v)", message.Watermark)
 
 	// Start to write an entry again
 	err = newWal.Write(&message)
diff --git a/test/redis-source-e2e/redis_source_test.go b/test/redis-streams-e2e/redis_streams_test.go
similarity index 95%
rename from test/redis-source-e2e/redis_source_test.go
rename to test/redis-streams-e2e/redis_streams_test.go
index 9758d503..e9e58bb6 100644
--- a/test/redis-source-e2e/redis_source_test.go
+++ b/test/redis-streams-e2e/redis_streams_test.go
@@ -16,7 +16,7 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package redis_source_e2e
+package redis_streams_e2e
 
 import (
 	"encoding/json"
@@ -28,7 +28,6 @@ import (
 	"github.com/numaproj/numaflow/test/fixtures"
 )
 
-//go:generate kubectl -n numaflow-system delete statefulset nats --ignore-not-found=true
 //go:generate kubectl apply -k ../../config/apps/redis -n numaflow-system
 type RedisSourceSuite struct {
 	fixtures.E2ESuite