Added device update event for hello

Author: MaxThom

TASKS.md

@@ -24,11 +24,16 @@
     - [x] BUG why invalid wired format
       - [x] Add logger to mir sdk
       - [x] Deploy
-    - [ ] Perf Improvements
+    - [x] Perf Improvements
+      - [x] protocfg/cmd schema protocache
+        - [x] need an event on device hello for schema
+              1. new schema update event maybe? no just use update
+              2. on hello received or on the buffer flush? buffer flush
+        - [x] change protocache to accept new schema even if not request before so its ready
       - [x] Test hearthbeat with hello
-      - [ ] prototlm/protocfg, if device dont exist, dont ask schema
-      - [ ] Bulk create devices
-        - [ ] Use for swarm
+      - [x] prototlm/protocfg, if device dont exist, dont ask schema
+      - [x] Bulk create devices
+        - [x] Use for swarm
       - [x] Stack Events and flush them every x seconds
         - [x] With many create at the same time
       - [x] Stack Hearbeaths and flush them every x seconds

internal/servers/core_srv/server.go

@@ -398,7 +398,6 @@ func (s *CoreServer) hearthbeatOnlinePulsor() {
 	s.hearthbeatsWriteBufferMu.Unlock()
 
 	degradedMode := false
-
 	devsResp, err := retry.RetryOnErrorContainsWithResult(func() ([]mir_v1.Device, error) {
 		res, err := s.store.UpdateDeviceHello(tempBuffer)
 		return res, err
@@ -418,7 +417,7 @@ func (s *CoreServer) hearthbeatOnlinePulsor() {
 			}).WithStatus(mir_v1.DeviceStatus{
 				Online:         &degradedMode,
 				LastHearthbeat: &surrealdbModels.CustomDateTime{Time: t.Hearthbeat},
-			})
+			}).WithSchema(t.Schema, t.Hearthbeat)
 			i += 1
 		}
 	}
@@ -471,6 +470,12 @@ func (s *CoreServer) hearthbeatOnlinePulsor() {
 			newOnline = append(newOnline, dev.Spec.DeviceId)
 		}
 		s.hearthbeats[mir_v1.DeviceId(dev.Spec.DeviceId)] = dev.Status.LastHearthbeat.Time
+
+		if hello, ok := tempBuffer[mir_v1.DeviceId(dev.Spec.DeviceId)]; ok && hello.Schema != nil {
+			if err := publishDeviceUpdateEvent(s.m, nil, dev); err != nil {
+				l.Warn().Err(err).Str("device_id", dev.Spec.DeviceId).Msg("error occure while publishing device update event")
+			}
+		}
 	}
 	s.hearthbeatsMutex.Unlock()
 

internal/servers/core_srv/server_integration_test.go

@@ -2513,6 +2513,17 @@ func TestDeviceAutoProvisionWithSchema(t *testing.T) {
 			msg.Ack()
 		})
 
+	// Subscribe to device update event
+	updEvent := make(chan bool, 1)
+	updateEventCount := 0
+	u, err := mSdk.Bus.Subscribe(
+		core_client.DeviceUpdatedEvent.WithId(deviceIds[0]),
+		func(msg *nats.Msg) {
+			updateEventCount += 1
+			updEvent <- true
+			msg.Ack()
+		})
+
 	// Act
 	fSet := &descriptorpb.FileDescriptorSet{
 		File: []*descriptorpb.FileDescriptorProto{
@@ -2526,11 +2537,19 @@ func TestDeviceAutoProvisionWithSchema(t *testing.T) {
 		t.Error(err)
 	}
 
-	select {
-	case <-msgReceived:
-		// Device created event received
-	case <-time.After(20 * time.Second):
-		t.Error("Timeout waiting for device auto-provision")
+	count := 0
+EvtLoop:
+	for count < 2 {
+		select {
+		case <-msgReceived:
+			// Device created event received
+			count++
+		case <-updEvent:
+			count++
+		case <-time.After(20 * time.Second):
+			t.Error("Timeout waiting for device auto-provision")
+			break EvtLoop
+		}
 	}
 
 	respList, err := core_client.PublishDeviceListRequest(mSdk.Bus, reqList)
@@ -2551,7 +2570,9 @@ func TestDeviceAutoProvisionWithSchema(t *testing.T) {
 	}
 	assert.Equal(t, 1, onlineEventCount)
 	assert.Equal(t, 1, createEventCount)
+	assert.Equal(t, 1, updateEventCount)
 	s.Unsubscribe()
+	u.Unsubscribe()
 	c.Unsubscribe()
 }
 

internal/services/schema_cache/proto_cache.go

@@ -241,13 +241,13 @@ func (c *MirSchemaCache) deviceUpdateSub(msg *mir.Msg, deviceId string, device m
 
 	// We dont update the cache with new elements.
 	// It has to be requested first
-	c.cacheLock.RLock()
-	if _, ok := c.cache[deviceId]; !ok {
-		msg.Ack()
-		c.cacheLock.RUnlock()
-		return
-	}
-	c.cacheLock.RUnlock()
+	// c.cacheLock.RLock()
+	// if _, ok := c.cache[deviceId]; !ok {
+	// 	msg.Ack()
+	// 	c.cacheLock.RUnlock()
+	// 	return
+	// }
+	// c.cacheLock.RUnlock()
 
 	if err != nil {
 		l.Error().Str("device_id", deviceId).Err(err).Msg("error deserializing event")

internal/services/schema_cache/proto_cache_integration_test.go

@@ -143,7 +143,7 @@ func TestPublishDeviceUpdateCache(t *testing.T) {
 	// Assert
 	assert.Equal(t, true, mir_proto.AreSchemaEqual(ogSch, sch))
 	assert.Equal(t, devPostUpd.Meta.Labels["test"], str)
-	assert.Equal(t, 1, count)
+	assert.Equal(t, 2, count)
 	cancel()
 	wg.Wait()
 }

pkgs/mir_v1/device.go

@@ -74,6 +74,15 @@ func (d Device) WithStatus(s DeviceStatus) Device {
 	return d
 }
 
+func (d Device) WithSchema(s *mir_proto.MirProtoSchema, t time.Time) Device {
+	if s == nil {
+		return d
+	}
+	d.Status.Schema, _ = NewSchemaFromProtoSchema(s)
+	d.Status.Schema.LastSchemaFetch = &surrealdbModels.CustomDateTime{Time: t}
+	return d
+}
+
 func (d Device) GetNameNamespace() string {
 	return d.Meta.Name + "/" + d.Meta.Namespace
 }