struct migration to support storing events in gomemdb

nomad/event_endpoint.go

@@ -83,7 +83,7 @@ func (e *Event) stream(conn io.ReadWriteCloser) {
 	}
 	defer subscription.Unsubscribe()
 
-	ndJsonCh := make(chan *stream.NDJson)
+	ndJsonCh := make(chan *structs.NDJson)
 	errCh := make(chan error)
 
 	jsonStream := stream.NewNDJsonStream(ndJsonCh, 30*time.Second)

nomad/state/state_changes.go

@@ -3,7 +3,6 @@ package state
 import (
 	"fmt"
 
-	"github.com/davecgh/go-spew/spew"
 	"github.com/hashicorp/go-memdb"
 	"github.com/hashicorp/nomad/nomad/stream"
 	"github.com/hashicorp/nomad/nomad/structs"
@@ -175,7 +174,6 @@ func (tx *txn) Commit() error {
 		}
 
 		if tx.persistChanges {
-			spew.Dump("YO!")
 			// persist events after processing changes
 			err := tx.Txn.Insert("events", events)
 			if err != nil {

nomad/stream/ndjson.go

@@ -7,43 +7,40 @@ import (
 	"fmt"
 	"sync"
 	"time"
+
+	"github.com/hashicorp/nomad/nomad/structs"
 )
 
 var (
 	// NDJsonHeartbeat is the NDJson to send as a heartbeat
 	// Avoids creating many heartbeat instances
-	NDJsonHeartbeat = &NDJson{Data: []byte("{}\n")}
+	NDJsonHeartbeat = &structs.NDJson{Data: []byte("{}\n")}
 )
 
 // NDJsonStream is used to send new line delimited JSON and heartbeats
 // to a destination (out channel)
 type NDJsonStream struct {
-	out chan<- *NDJson
+	out chan<- *structs.NDJson
 
 	// heartbeat is the interval to send heartbeat messages to keep a connection
 	// open.
 	heartbeat *time.Ticker
 
-	publishCh chan NDJson
+	publishCh chan structs.NDJson
 	exitCh    chan struct{}
 
 	l       sync.Mutex
 	running bool
 }
 
-// NNDJson is a wrapper for a Newline Delimited JSON object
-type NDJson struct {
-	Data []byte
-}
-
 // NewNNewNDJsonStream creates a new NDJson stream that will output NDJson structs
 // to the passed output channel
-func NewNDJsonStream(out chan<- *NDJson, heartbeat time.Duration) *NDJsonStream {
+func NewNDJsonStream(out chan<- *structs.NDJson, heartbeat time.Duration) *NDJsonStream {
 	return &NDJsonStream{
 		out:       out,
 		heartbeat: time.NewTicker(heartbeat),
 		exitCh:    make(chan struct{}),
-		publishCh: make(chan NDJson),
+		publishCh: make(chan structs.NDJson),
 	}
 }
 
@@ -97,18 +94,10 @@ func (n *NDJsonStream) Send(obj interface{}) error {
 	}
 
 	select {
-	case n.publishCh <- NDJson{Data: buf.Bytes()}:
+	case n.publishCh <- structs.NDJson{Data: buf.Bytes()}:
 	case <-n.exitCh:
 		return fmt.Errorf("stream is no longer running")
 	}
 
 	return nil
 }
-
-func (j *NDJson) Copy() *NDJson {
-	n := new(NDJson)
-	*n = *j
-	n.Data = make([]byte, len(j.Data))
-	copy(n.Data, j.Data)
-	return n
-}

nomad/stream/ndjson_test.go

@@ -6,6 +6,7 @@ import (
 	"testing"
 	"time"
 
+	"github.com/hashicorp/nomad/nomad/structs"
 	"github.com/stretchr/testify/require"
 )
 
@@ -19,7 +20,7 @@ func TestNDJson(t *testing.T) {
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
 
-	out := make(chan *NDJson)
+	out := make(chan *structs.NDJson)
 	s := NewNDJsonStream(out, 1*time.Second)
 	s.Run(ctx)
 
@@ -45,7 +46,7 @@ func TestNDJson_Send_After_Stop(t *testing.T) {
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
 
-	out := make(chan *NDJson)
+	out := make(chan *structs.NDJson)
 	s := NewNDJsonStream(out, 1*time.Second)
 	s.Run(ctx)
 
@@ -62,7 +63,7 @@ func TestNDJson_HeartBeat(t *testing.T) {
 	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
 
-	out := make(chan *NDJson)
+	out := make(chan *structs.NDJson)
 	s := NewNDJsonStream(out, 10*time.Millisecond)
 	s.Run(ctx)
 

nomad/structs/structs.go

@@ -39,7 +39,6 @@ import (
 	"github.com/hashicorp/nomad/helper/constraints/semver"
 	"github.com/hashicorp/nomad/helper/uuid"
 	"github.com/hashicorp/nomad/lib/kheap"
-	"github.com/hashicorp/nomad/nomad/stream"
 	psstructs "github.com/hashicorp/nomad/plugins/shared/structs"
 )
 
@@ -10679,7 +10678,7 @@ type EventStreamRequest struct {
 
 type EventStreamWrapper struct {
 	Error *RpcError
-	Event *stream.NDJson
+	Event *NDJson
 }
 
 // RpcError is used for serializing errors with a potential error code
@@ -10700,6 +10699,7 @@ func (r *RpcError) Error() string {
 }
 
 type Topic string
+
 type Event struct {
 	Topic      Topic
 	Type       string
@@ -10714,18 +10714,15 @@ type Events struct {
 	Events []Event
 }
 
-type Topic string
-
-type Event struct {
-	Topic      Topic
-	Type       string
-	Key        string
-	FilterKeys []string
-	Index      uint64
-	Payload    interface{}
+// NNDJson is a wrapper for a Newline Delimited JSON object
+type NDJson struct {
+	Data []byte
 }
 
-type Events struct {
-	Index  uint64
-	Events []Event
+func (j *NDJson) Copy() *NDJson {
+	n := new(NDJson)
+	*n = *j
+	n.Data = make([]byte, len(j.Data))
+	copy(n.Data, j.Data)
+	return n
 }