Skip to content

Commit 44e1bcd

Browse files
authored
feat: store generated changeset into ingestion log (#172)
1 parent 4be8d62 commit 44e1bcd

File tree

7 files changed

+508
-145
lines changed

7 files changed

+508
-145
lines changed

diode-proto/diode/v1/reconciler.proto

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -70,6 +70,12 @@ message IngestionMetrics {
7070
int32 no_changes = 5;
7171
}
7272

73+
// A change set
74+
message ChangeSet {
75+
string id = 1; // A change set ID
76+
bytes data = 2; // Binary data representing the change set
77+
}
78+
7379
// An ingestion log
7480
message IngestionLog {
7581
string id = 1;
@@ -83,6 +89,7 @@ message IngestionLog {
8389
string sdk_version = 9;
8490
diode.v1.Entity entity = 10;
8591
IngestionError error = 11;
92+
ChangeSet change_set = 12;
8693
}
8794

8895
// The request to retrieve ingestion logs

diode-server/gen/diode/v1/reconcilerpb/reconciler.pb.go

Lines changed: 215 additions & 131 deletions
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.

diode-server/gen/diode/v1/reconcilerpb/reconciler.pb.validate.go

Lines changed: 132 additions & 0 deletions
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.

diode-server/go.mod

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -23,6 +23,7 @@ require (
2323

2424
require (
2525
github.com/alicebob/gopher-json v0.0.0-20200520072559-a9ecdc9d1d3a // indirect
26+
github.com/andybalholm/brotli v1.1.0
2627
github.com/cespare/xxhash/v2 v2.2.0 // indirect
2728
github.com/davecgh/go-spew v1.1.1 // indirect
2829
github.com/dgryski/go-rendezvous v0.0.0-20200823014737-9f7001d12a5f // indirect

diode-server/go.sum

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2,6 +2,8 @@ github.com/alicebob/gopher-json v0.0.0-20200520072559-a9ecdc9d1d3a h1:HbKu58rmZp
22
github.com/alicebob/gopher-json v0.0.0-20200520072559-a9ecdc9d1d3a/go.mod h1:SGnFV6hVsYE877CKEZ6tDNTjaSXYUk6QqoIK6PrAtcc=
33
github.com/alicebob/miniredis/v2 v2.33.0 h1:uvTF0EDeu9RLnUEG27Db5I68ESoIxTiXbNUiji6lZrA=
44
github.com/alicebob/miniredis/v2 v2.33.0/go.mod h1:MhP4a3EU7aENRi9aO+tHfTBZicLqQevyi/DJpoj6mi0=
5+
github.com/andybalholm/brotli v1.1.0 h1:eLKJA0d02Lf0mVpIDgYnqXcUn0GqVmEFny3VuID1U3M=
6+
github.com/andybalholm/brotli v1.1.0/go.mod h1:sms7XGricyQI9K10gOSf56VKKWS4oLer58Q+mhRPtnY=
57
github.com/bsm/ginkgo/v2 v2.12.0 h1:Ny8MWAHyOepLGlLKYmXG4IEkioBysk6GpaRTLC8zwWs=
68
github.com/bsm/ginkgo/v2 v2.12.0/go.mod h1:SwYbGRRDovPVboqFv0tPTcG1sN61LM1Z4ARdbAV9g4c=
79
github.com/bsm/gomega v1.27.10 h1:yeMWxP2pV2fG3FgAODIY8EiRE3dy0aeFYt4l7wh6yKA=

diode-server/reconciler/ingestion_processor.go

Lines changed: 39 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -1,14 +1,17 @@
11
package reconciler
22

33
import (
4+
"bytes"
45
"context"
6+
"encoding/json"
57
"errors"
68
"fmt"
79
"log/slog"
810
"os"
911
"regexp"
1012
"strconv"
1113

14+
"github.com/andybalholm/brotli"
1215
"github.com/kelseyhightower/envconfig"
1316
"github.com/redis/go-redis/v9"
1417
"github.com/segmentio/ksuid"
@@ -232,6 +235,16 @@ func (p *IngestionProcessor) handleStreamMessage(ctx context.Context, msg redis.
232235
ingestionLog.State = reconcilerpb.State_FAILED
233236
ingestionLog.Error = extractIngestionError(err)
234237

238+
if changeSet != nil {
239+
ingestionLog.ChangeSet = &reconcilerpb.ChangeSet{Id: changeSet.ChangeSetID}
240+
csCompressed, err := compressChangeSet(changeSet)
241+
if err != nil {
242+
errs = append(errs, err)
243+
} else {
244+
ingestionLog.ChangeSet.Data = csCompressed
245+
}
246+
}
247+
235248
if _, err = p.writeIngestionLog(ctx, key, ingestionLog); err != nil {
236249
errs = append(errs, err)
237250
}
@@ -240,7 +253,13 @@ func (p *IngestionProcessor) handleStreamMessage(ctx context.Context, msg redis.
240253

241254
if changeSet != nil {
242255
ingestionLog.State = reconcilerpb.State_RECONCILED
243-
//TODO: add change set ID to ingestion log
256+
ingestionLog.ChangeSet = &reconcilerpb.ChangeSet{Id: changeSet.ChangeSetID}
257+
csCompressed, err := compressChangeSet(changeSet)
258+
if err != nil {
259+
errs = append(errs, err)
260+
} else {
261+
ingestionLog.ChangeSet.Data = csCompressed
262+
}
244263
} else {
245264
ingestionLog.State = reconcilerpb.State_NO_CHANGES
246265
}
@@ -328,7 +347,7 @@ func (p *IngestionProcessor) reconcileEntity(ctx context.Context, ingestEntity c
328347

329348
resp, err := p.nbClient.ApplyChangeSet(ctx, req)
330349
if err != nil {
331-
return nil, err
350+
return cs, err
332351
}
333352

334353
p.logger.Debug("apply change set response", "response", resp)
@@ -356,6 +375,24 @@ func normalizeIngestionLog(l []byte) []byte {
356375
return re.ReplaceAll(l, []byte(`"ingestionTs":$1`))
357376
}
358377

378+
func compressChangeSet(cs *changeset.ChangeSet) ([]byte, error) {
379+
csJSON, err := json.Marshal(cs)
380+
if err != nil {
381+
return nil, fmt.Errorf("failed to marshal changeset JSON: %v", err)
382+
}
383+
384+
var brotliBuf bytes.Buffer
385+
brotliWriter := brotli.NewWriter(&brotliBuf)
386+
if _, err = brotliWriter.Write(csJSON); err != nil {
387+
return nil, fmt.Errorf("failed to compress changeset: %v", err)
388+
}
389+
if err = brotliWriter.Close(); err != nil {
390+
return nil, fmt.Errorf("failed to compress changeset: %v", err)
391+
}
392+
393+
return brotliBuf.Bytes(), nil
394+
}
395+
359396
func extractObjectType(in *diodepb.Entity) (string, error) {
360397
switch in.GetEntity().(type) {
361398
case *diodepb.Entity_Device:

0 commit comments

Comments
 (0)