Merge pull request #141 from thin-edge/refactor-entity-store-api

refactor: use tedge entity store api for registration

Author: reubenmiller

.goreleaser.yaml

@@ -85,6 +85,17 @@ nfpms:
       - rpm
       - apk
 
+    overrides:
+      # Dependency on the Entity HTTP API which was only added in tedge >= 1.5.0
+      # Note: Don't add explicit dependency for apk (Alpine Linux)
+      # as generally only the tedge is just installed via a binary and not the apk package
+      deb:
+        dependencies:
+          - tedge (>= 1.5.0)
+      rpm:
+        dependencies:
+          - tedge >= 1.5.0-1
+
     # FIXME: Remove for official release, as the package can be called "tedge-container-plugin" instead of "tedge-container-plugin-ng"
     replaces:
       - tedge-container-plugin

cli/run/cmd.go

@@ -50,6 +50,8 @@ func NewRunCommand(cliContext cli.Cli) *cobra.Command {
 				DeleteFromCloud:    cliContext.DeleteFromCloud(),
 				EnableEngineEvents: cliContext.EngineEventsEnabled(),
 
+				HTTPHost:       cliContext.GetHTTPHost(),
+				HTTPPort:       cliContext.GetHTTPPort(),
 				MQTTHost:       cliContext.GetMQTTHost(),
 				MQTTPort:       cliContext.GetMQTTPort(),
 				CumulocityHost: cliContext.GetCumulocityHost(),
@@ -158,6 +160,8 @@ func NewRunCommand(cliContext cli.Cli) *cobra.Command {
 	viper.SetDefault("delete_from_cloud.enabled", true)
 
 	// thin-edge.io services
+	viper.SetDefault("client.http.host", "127.0.0.1")
+	viper.SetDefault("client.http.port", 8000)
 	viper.SetDefault("client.mqtt.host", "127.0.0.1")
 	// client.mqtt.port: 0 = auto-detection, where 8883 is used when the cert files exist, or 1883 otherwise
 	viper.SetDefault("client.mqtt.port", 0)

packaging/config.toml

@@ -30,6 +30,11 @@ labels = [ "tedge.ignore" ]
 key = "/etc/tedge/device-certs/local-tedge.key"
 # Certificate file used to connect to local thin-edge.io services (only used if the file exists)
 cert_file = "/etc/tedge/device-certs/local-tedge.crt"
+  [client.http]
+  # thin-edge.io HTTP host operated by the tedge-agent on the main device
+  host = "127.0.0.1"
+  # thin-edge.io HTTP port operated by the tedge-agent on the main device
+  port = 8000
 
   [client.mqtt]
   # thin-edge.io MQTT host

pkg/app/app.go

@@ -69,6 +69,9 @@ type Config struct {
 	EnableEngineEvents bool
 	DeleteFromCloud    bool
 
+	HTTPHost string
+	HTTPPort uint16
+
 	MQTTHost string
 	MQTTPort uint16
 
@@ -79,6 +82,8 @@ type Config struct {
 func NewApp(device tedge.Target, config Config) (*App, error) {
 	serviceTarget := device.Service(config.ServiceName)
 	tedgeOpts := &tedge.ClientConfig{
+		HTTPHost: config.HTTPHost,
+		HTTPPort: config.HTTPPort,
 		MqttHost: config.MQTTHost,
 		MqttPort: config.MQTTPort,
 		C8yHost:  config.CumulocityHost,
@@ -94,6 +99,18 @@ func NewApp(device tedge.Target, config Config) (*App, error) {
 		return nil, err
 	}
 
+	// Register via http interface
+	_, registrationErr := tedgeClient.TedgeAPI.CreateEntity(context.Background(), tedge.Entity{
+		TedgeType:    tedge.EntityTypeService,
+		Name:         serviceTarget.Name,
+		TedgeTopicID: serviceTarget.TopicID,
+	})
+	if registrationErr == nil {
+		slog.Info("Registered service", "topic", serviceTarget.Topic())
+	} else {
+		slog.Error("Could not register tedge entity.", "err", registrationErr)
+	}
+
 	if err := tedgeClient.Connect(); err != nil {
 		return nil, err
 	}
@@ -137,13 +154,10 @@ func (a *App) DeleteLegacyService(deleteFromCloud bool) {
 	target := a.client.Target.Service("tedge-container-monitor")
 	slog.Info("Removing legacy service from the cloud", "topic", target.Topic())
 
-	if err := a.client.Publish(tedge.GetHealthTopic(*target), 1, true, ""); err != nil {
-		slog.Warn("Failed to clear health status.", "topic", tedge.GetHealthTopic(*target))
-	}
-	time.Sleep(500 * time.Millisecond)
-	if err := a.client.Publish(tedge.GetTopic(*target), 1, true, ""); err != nil {
-		slog.Warn("Failed to clear registration message.", "topic", tedge.GetHealthTopic(*target))
+	if _, err := a.client.TedgeAPI.DeleteEntity(context.Background(), *target); err != nil {
+		slog.Warn("Failed to clear registration message.", "topic-id", target.TopicID)
 	}
+
 	time.Sleep(500 * time.Millisecond)
 
 	if target.CloudIdentity != "" && deleteFromCloud {
@@ -373,10 +387,16 @@ func (a *App) updateMetrics(items []container.TedgeContainer) error {
 	doWork := func(jobs <-chan container.TedgeContainer, results chan<- error) {
 		for j := range jobs {
 			var jobErr error
-			stats, jobErr := a.ContainerClient.GetStats(context.Background(), j.Container.Id)
+			// TODO: Check if container exists, if not then do nothing
+			target := a.Device.Service(j.Name)
+			if _, entityErr := a.client.TedgeAPI.GetEntity(context.Background(), *target); entityErr != nil {
+				slog.Info("Entity has not be registered yet, skipping metric for it.", "topic-id", target.TopicID)
+				results <- jobErr
+				continue
+			}
 
+			stats, jobErr := a.ContainerClient.GetStats(context.Background(), j.Container.Id)
 			if jobErr == nil {
-				target := a.Device.Service(j.Name)
 				topic := tedge.GetTopic(*target, "m", "resource_usage")
 				payload, err := json.Marshal(stats)
 				if err == nil {
@@ -443,25 +463,26 @@ func (a *App) doUpdate(filterOptions container.FilterOptions) error {
 		target := a.Device.Service(item.Name)
 
 		// Skip registration message if it already exists
-		if _, ok := existingServices[target.Topic()]; ok {
-			slog.Debug("Container is already registered", "topic", target.Topic())
-			delete(existingServices, target.Topic())
-			continue
-		}
+		// if _, ok := existingServices[target.Topic()]; ok {
+		// 	slog.Info("Container is already registered", "topic", target.Topic())
+		// 	delete(existingServices, target.Topic())
+		// 	continue
+		// }
 		delete(existingServices, target.Topic())
 
-		payload := map[string]any{
-			"@type": "service",
-			"name":  item.Name,
-			"type":  item.ServiceType,
-		}
-		b, err := json.Marshal(payload)
-		if err != nil {
-			slog.Warn("Could not marshal registration message", "err", err)
-			continue
-		}
-		if err := tedgeClient.Publish(target.Topic(), 1, true, b); err != nil {
-			slog.Error("Failed to register container", "target", target.Topic(), "err", err)
+		// Register using HTTP API
+		resp, err := tedgeClient.TedgeAPI.CreateEntity(context.Background(), tedge.Entity{
+			TedgeType:     tedge.EntityTypeService,
+			TedgeTopicID:  target.TopicID,
+			Name:          item.Name,
+			Type:          item.ServiceType,
+			TedgeParentID: tedgeClient.Parent.TopicID,
+		})
+
+		if err == nil {
+			slog.Info("Registered container.", "topic", target.Topic(), "url", resp.RawResponse.Request.URL.String(), "status_code", resp.RawResponse.Status)
+		} else {
+			slog.Error("Failed to register container.", "topic", target.Topic(), "err", err)
 		}
 	}
 
@@ -490,18 +511,16 @@ func (a *App) doUpdate(filterOptions container.FilterOptions) error {
 	for _, item := range items {
 		target := a.Device.Service(item.Name)
 
-		topic := tedge.GetTopic(*target, "twin", "container")
-
 		// Create status
-		payload, err := json.Marshal(item.Container)
-
+		_, err := tedgeClient.TedgeAPI.UpdateTwin(
+			context.Background(),
+			tedge.Entity{
+				TedgeTopicID: target.TopicID,
+			},
+			"container",
+			item.Container,
+		)
 		if err != nil {
-			slog.Error("Failed to convert payload to json", "err", err)
-			continue
-		}
-
-		slog.Info("Publishing container status", "topic", topic, "payload", payload)
-		if err := tedgeClient.Publish(topic, 1, true, payload); err != nil {
 			slog.Error("Could not publish container status", "err", err)
 		}
 	}
@@ -518,7 +537,7 @@ func (a *App) doUpdate(filterOptions container.FilterOptions) error {
 				continue
 			}
 
-			if err := tedgeClient.DeregisterEntity(*target, "twin/container"); err != nil {
+			if err := tedgeClient.DeregisterEntity(*target); err != nil {
 				slog.Warn("Failed to deregister entity.", "err", err)
 			}
 

pkg/cli/cli.go

@@ -155,6 +155,18 @@ func (c *Cli) DeleteFromCloud() bool {
 	return viper.GetBool("delete_from_cloud.enabled")
 }
 
+func (c *Cli) GetHTTPHost() string {
+	return viper.GetString("client.http.host")
+}
+
+func (c *Cli) GetHTTPPort() uint16 {
+	v := viper.GetUint16("client.http.port")
+	if v == 0 {
+		return 8000
+	}
+	return v
+}
+
 func (c *Cli) GetMQTTHost() string {
 	return viper.GetString("client.mqtt.host")
 }

pkg/tedge/target.go

@@ -5,10 +5,14 @@ import (
 	"strings"
 )
 
+var EntityTypeService = "service"
+var EntityTypeChildDevice = "child-device"
+
 type Target struct {
 	RootPrefix    string
 	TopicID       string
 	CloudIdentity string
+	Name          string
 }
 
 func (t *Target) ExternalID() string {
@@ -25,6 +29,7 @@ func (t *Target) Topic() string {
 func (t *Target) Service(name string) *Target {
 	target := NewTarget(t.RootPrefix, strings.Join(strings.Split(t.TopicID, "/")[0:2], "/")+"/service/"+name)
 	target.CloudIdentity = t.CloudIdentity
+	target.Name = name
 	return target
 }