Merge pull request #451 from kcajmagic/feature/sessionID

add session-id to event metadata

CHANGELOG.md

@@ -5,6 +5,8 @@ The format is based on [Keep a Changelog](http://keepachangelog.com/en/1.0.0/)
 and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.html).
 
 ## [Unreleased]
+- Added session-id to device information [#451](https://github.com/xmidt-org/webpa-common/pull/451)
+- Added device metadata to outbound events [#451](https://github.com/xmidt-org/webpa-common/pull/451)
 
 ## [v1.5.1]
 - Added automated releases using travis [#444](https://github.com/xmidt-org/webpa-common/pull/444)

device/device.go

@@ -2,6 +2,8 @@ package device
 
 import (
 	"bytes"
+	"crypto/rand"
+	"encoding/base64"
 	"encoding/json"
 	"fmt"
 	"sync/atomic"
@@ -96,6 +98,9 @@ type Interface interface {
 	// SatClientID returns the SAT JWT token passed when the device connected
 	SatClientID() string
 
+	// SessionID returns the session id associated with the devices connection
+	SessionID() string
+
 	// Trust returns the trust level of this device
 	Trust() string
 
@@ -127,6 +132,7 @@ type device struct {
 
 	partnerIDs  []string
 	satClientID string
+	sessionID   string
 
 	trust string
 
@@ -166,6 +172,10 @@ func newDevice(o deviceOptions) *device {
 	var partnerIDs []string
 	partnerIDs = append(partnerIDs, o.PartnerIDs...)
 
+	var id [16]byte
+	rand.Read(id[:])
+	sessionID := base64.RawURLEncoding.EncodeToString(id[:])
+
 	return &device{
 		id:           o.ID,
 		errorLog:     logging.Error(o.Logger, "id", o.ID),
@@ -180,6 +190,7 @@ func newDevice(o deviceOptions) *device {
 		transactions: NewTransactions(),
 		partnerIDs:   partnerIDs,
 		satClientID:  o.SatClientID,
+		sessionID:    sessionID,
 		trust:        o.Trust,
 	}
 }
@@ -337,6 +348,10 @@ func (d *device) SatClientID() string {
 	return d.satClientID
 }
 
+func (d *device) SessionID() string {
+	return d.sessionID
+}
+
 func (d *device) Trust() string {
 	return d.trust
 }

device/device_test.go

@@ -57,6 +57,7 @@ func TestDevice(t *testing.T) {
 		)
 
 		require.NotNil(device)
+		assert.NotEmpty(device.sessionID)
 		device.statistics = NewStatistics(func() time.Time { return expectedConnectedAt.Add(expectedUpTime) }, expectedConnectedAt)
 
 		assert.Equal(string(record.expectedID), device.String())
@@ -104,3 +105,19 @@ func TestDevice(t *testing.T) {
 		assert.Error(err)
 	}
 }
+
+func TestDeviceSessionID(t *testing.T) {
+	assert := assert.New(t)
+
+	connectOptions := deviceOptions{
+		ID:          "1",
+		QueueSize:   10,
+		ConnectedAt: time.Now(),
+		Logger:      logging.NewTestLogger(nil, t),
+	}
+	sessionOne := newDevice(connectOptions)
+	sessionTwo := newDevice(connectOptions)
+	assert.Equal(sessionOne.ID(), sessionTwo.ID())
+	assert.NotEqual(sessionOne.SessionID(), sessionTwo.SessionID())
+}
+

device/manager.go

@@ -4,6 +4,7 @@ import (
 	"encoding/json"
 	"io"
 	"net/http"
+	"strings"
 	"sync"
 	"time"
 
@@ -340,7 +341,7 @@ func (m *manager) readPump(d *device, r ReadCloser, closeOnce *sync.Once) {
 				event.Type = TransactionComplete
 			}
 		}
-
+		event = updateEventMetadata(event)
 		m.dispatch(&event)
 	}
 }
@@ -490,3 +491,28 @@ func (m *manager) Route(request *Request) (*Response, error) {
 		return nil, ErrorDeviceNotFound
 	}
 }
+
+func updateEventMetadata(event Event) Event {
+	var (
+		contents []byte
+		msg      wrp.Message
+	)
+
+	// decode event
+	err := wrp.NewDecoderBytes(event.Contents, event.Format).Decode(&msg)
+	if err != nil {
+		return event
+	}
+
+	// update metadata
+	msg.Metadata["partner-ids"] = strings.Join(event.Device.PartnerIDs(), ",")
+	msg.Metadata["session-id"] = event.Device.SessionID()
+
+	// encode event
+	err = wrp.NewEncoderBytes(&contents, event.Format).Encode(&msg)
+	if err != nil {
+		return event
+	}
+	event.Contents = contents
+	return event
+}

device/manager_test.go

@@ -6,6 +6,7 @@ import (
 	"net/http"
 	"net/http/httptest"
 	"net/url"
+	"strings"
 	"sync"
 	"testing"
 	"time"
@@ -395,3 +396,54 @@ func TestGaugeCardinality(t *testing.T) {
 		m.(*manager).measures.Models.With("neat", "bad").Add(-1)
 	})
 }
+
+func TestUpdateMetadata(t *testing.T) {
+	assert := assert.New(t)
+	require := require.New(t)
+	msg := wrp.Message{
+		Type:            wrp.SimpleEventMessageType,
+		Source:          "external.com",
+		Destination:     "mac:FFEEAADD44443333",
+		TransactionUUID: "DEADBEEF",
+		Headers:         []string{"Header1", "Header2"},
+		Metadata:        map[string]string{"name": "value"},
+		Spans:           [][]string{{"1", "2"}, {"3"}},
+		Payload:         []byte{1, 2, 3, 4, 0xff, 0xce},
+		PartnerIDs:      []string{"foo"},
+	}
+	originalContents := []byte{}
+	err := wrp.NewEncoderBytes(&originalContents, wrp.Msgpack).Encode(&msg)
+	require.NoError(err)
+	device := newDevice(deviceOptions{
+		ID:          "mac:FFEEAADD44443333",
+		C:           nil,
+		Compliance:  0,
+		PartnerIDs:  nil,
+		SatClientID: "",
+		Trust:       "",
+		QueueSize:   0,
+		ConnectedAt: time.Time{},
+		Logger:      nil,
+	})
+	originalEvent := Event{
+		Type:     MessageReceived,
+		Device:   device,
+		Message:  &msg,
+		Format:   wrp.Msgpack,
+		Contents: originalContents,
+	}
+
+	event := updateEventMetadata(originalEvent)
+
+	assert.Equal(originalEvent.Device, event.Device)
+	assert.Equal(originalEvent.Message, event.Message)
+	assert.NotEqual(originalEvent.Contents, event.Contents)
+	assert.Equal(originalEvent.Format, event.Format)
+
+	expectedMetaData := map[string]string{"name": "value", "session-id": device.SessionID(), "partner-ids": strings.Join(device.PartnerIDs(), ",")}
+
+	newMessage := new(wrp.Message)
+	err = wrp.NewDecoderBytes(event.Contents, event.Format).Decode(newMessage)
+	require.NoError(err)
+	assert.Equal(expectedMetaData, newMessage.Metadata)
+}

device/mocks.go

@@ -111,6 +111,12 @@ func (m *MockDevice) SatClientID() string {
 	return first
 }
 
+func (m *MockDevice) SessionID() string {
+	arguments := m.Called()
+	first, _ := arguments.Get(0).(string)
+	return first
+}
+
 func (m *MockDevice) Trust() string {
 	return m.Called().String(0)
 }