as: Add end device registry cache implementation

.golangci.yml

@@ -1,7 +1,3 @@
-run:
-  skip-dirs:
-    - node_modules
-
 linters:
   disable-all: true
   enable:
@@ -113,3 +109,6 @@ issues:
     - linters:
         - paralleltest
       text: 'does not use range value in test Run'
+  exclude-dirs:
+    - node_modules
+

CHANGELOG.md

@@ -11,10 +11,16 @@ For details about compatibility between different releases, see the **Commitment
 
 ### Added
 
+- Add the attributes field to all ApplicationUp messages.
+- Add the locations, version_ids, network_ids fields to ApplicationUp messages that were missing them, i.e. ApplicationJoinAccept, ApplicationDownlink, ApplicationDownlinkFailed, ApplicationInvalidatedDownlinks, ApplicationServiceData.
+- Add Timeout and Cache fields in the EndDeviceMetadataStorageConfig of the AS.
+
 ### Changed
 
 ### Deprecated
 
+- Deprecate the Location field (and its subfields) in the EndDeviceMetadataStorageConfig of AS.
+
 ### Removed
 
 ### Fixed

cmd/internal/shared/applicationserver/config.go

@@ -49,14 +49,12 @@ var DefaultApplicationServerConfig = applicationserver.Config{
 		Downlinks: web.DownlinksConfig{PublicAddress: shared.DefaultPublicURL + "/api/v3"},
 	},
 	EndDeviceMetadataStorage: applicationserver.EndDeviceMetadataStorageConfig{
-		Location: applicationserver.EndDeviceLocationStorageConfig{
-			Timeout: 5 * time.Second,
-			Cache: applicationserver.EndDeviceLocationStorageCacheConfig{
-				Enable:             true,
-				MinRefreshInterval: 15 * time.Minute,
-				MaxRefreshInterval: 4 * time.Hour,
-				TTL:                14 * 24 * time.Hour,
-			},
+		Timeout: 5 * time.Second,
+		Cache: applicationserver.EndDeviceRegistryStorageCacheConfig{
+			Enable:             true,
+			MinRefreshInterval: 15 * time.Minute,
+			MaxRefreshInterval: 4 * time.Hour,
+			TTL:                14 * 24 * time.Hour,
 		},
 	},
 	Distribution: applicationserver.DistributionConfig{

cmd/ttn-lw-stack/commands/start.go

@@ -445,21 +445,21 @@ var startCommand = &cobra.Command{
 				}
 				config.AS.Webhooks.Registry = webhookRegistry
 			}
-			if cache := &config.AS.EndDeviceMetadataStorage.Location.Cache; cache.Enable {
+			if cache := &config.AS.EndDeviceMetadataStorage.Cache; cache.Enable {
 				switch config.Cache.Service {
 				case "redis":
-					cache.Cache = &asmetaredis.EndDeviceLocationCache{
-						Redis: redis.New(config.Cache.Redis.WithNamespace("as", "metadata", "locations")),
+					cache.Cache = &asmetaredis.EndDeviceCache{
+						Redis: redis.New(config.Cache.Redis.WithNamespace("as", "metadata", "end-devices")),
 					}
 				default:
 					cache.Enable = false
 				}
 			}
-			locationRegistry, err := config.AS.EndDeviceMetadataStorage.Location.NewRegistry(ctx, c)
+			endDeviceRegistry, err := config.AS.EndDeviceMetadataStorage.NewRegistry(ctx, c)
 			if err != nil {
 				return shared.ErrInitializeApplicationServer.WithCause(err)
 			}
-			config.AS.EndDeviceMetadataStorage.Location.Registry = locationRegistry
+			config.AS.EndDeviceMetadataStorage.Registry = endDeviceRegistry
 			as, err := applicationserver.New(c, &config.AS)
 			if err != nil {
 				return shared.ErrInitializeApplicationServer.WithCause(err)

config/messages.json

@@ -2861,13 +2861,40 @@
       "file": "io.go"
     }
   },
+  "error:pkg/applicationserver/metadata/redis:cache_entry_malformed": {
+    "translations": {
+      "en": "cache entry malformed"
+    },
+    "description": {
+      "package": "pkg/applicationserver/metadata/redis",
+      "file": "redis.go"
+    }
+  },
   "error:pkg/applicationserver/metadata/redis:cache_miss": {
     "translations": {
       "en": "cache miss"
     },
     "description": {
       "package": "pkg/applicationserver/metadata/redis",
-      "file": "location_cache.go"
+      "file": "redis.go"
+    }
+  },
+  "error:pkg/applicationserver/metadata:end_device_not_found": {
+    "translations": {
+      "en": "end device not found"
+    },
+    "description": {
+      "package": "pkg/applicationserver/metadata",
+      "file": "metadata.go"
+    }
+  },
+  "error:pkg/applicationserver/metadata:field_mask_path_not_supported": {
+    "translations": {
+      "en": "field mask path `{path}` is not supported"
+    },
+    "description": {
+      "package": "pkg/applicationserver/metadata",
+      "file": "metadata.go"
     }
   },
   "error:pkg/applicationserver/redis:application_uid": {

pkg/applicationserver/applicationserver.go

@@ -81,7 +81,7 @@ type ApplicationServer struct {
 
 	linkRegistry           LinkRegistry
 	deviceRegistry         DeviceRegistry
-	locationRegistry       metadata.EndDeviceLocationRegistry
+	endDeviceRegistry      metadata.EndDeviceRegistry
 	formatters             messageprocessors.MapPayloadProcessor
 	webhooks               ioweb.Webhooks
 	webhookTemplates       ioweb.TemplateStore
@@ -156,14 +156,14 @@ func New(c *component.Component, conf *Config) (as *ApplicationServer, err error
 	}
 
 	as = &ApplicationServer{
-		Component:        c,
-		ctx:              ctx,
-		config:           conf,
-		linkRegistry:     conf.Links,
-		deviceRegistry:   wrapEndDeviceRegistryWithReplacedFields(conf.Devices, replacedEndDeviceFields...),
-		appPkgRegistry:   conf.Packages.Registry,
-		locationRegistry: conf.EndDeviceMetadataStorage.Location.Registry,
-		formatters:       make(messageprocessors.MapPayloadProcessor),
+		Component:         c,
+		ctx:               ctx,
+		config:            conf,
+		linkRegistry:      conf.Links,
+		deviceRegistry:    wrapEndDeviceRegistryWithReplacedFields(conf.Devices, replacedEndDeviceFields...),
+		appPkgRegistry:    conf.Packages.Registry,
+		endDeviceRegistry: conf.EndDeviceMetadataStorage.Registry,
+		formatters:        make(messageprocessors.MapPayloadProcessor),
 		clusterDistributor: distribution.NewPubSubDistributor(
 			ctx,
 			c,
@@ -1049,6 +1049,14 @@ func (as *ApplicationServer) handleJoinAccept(
 		return err
 	}
 
+	if entity, err := as.endDeviceRegistry.Get(ctx, ids, []string{"attributes"}); err != nil {
+		log.FromContext(ctx).WithError(err).Warn(
+			"Failed to retrieve end device attributes on join-accept",
+		)
+	} else {
+		joinAccept.Attributes = entity.Attributes
+	}
+
 	// Publish last seen event.
 	if err := as.deviceLastSeenPool.Publish(ctx, lastSeenAtInfo{
 		ids:        ids,
@@ -1159,6 +1167,7 @@ func (as *ApplicationServer) publishNormalizedUplink(ctx context.Context, info u
 					Locations:                 info.uplink.Locations,
 					VersionIds:                info.uplink.VersionIds,
 					NetworkIds:                info.uplink.NetworkIds,
+					Attributes:                info.uplink.Attributes,
 				},
 			},
 			Simulated: info.simulated,
@@ -1208,6 +1217,15 @@ func (as *ApplicationServer) handleUplink(ctx context.Context, info uplinkInfo)
 		return err
 	}
 
+	if entity, err := as.endDeviceRegistry.Get(ctx, info.ids, []string{"attributes", "locations"}); err != nil {
+		log.FromContext(ctx).WithError(err).Warn(
+			"Failed to retrieve the end device attributes and locations on uplink",
+		)
+	} else {
+		info.uplink.Attributes = entity.Attributes
+		info.uplink.Locations = entity.Locations
+	}
+
 	if !as.skipPayloadCrypto(ctx, info.link, dev, dev.Session) {
 		if err := as.decryptAndDecodeUplink(ctx, dev, info.uplink, info.link.DefaultFormatters); err != nil {
 			return err
@@ -1227,11 +1245,6 @@ func (as *ApplicationServer) handleUplink(ctx context.Context, info uplinkInfo)
 	}
 
 	// Set location in message and publish location solved if the payload contains location information.
-	if locations, err := as.locationRegistry.Get(ctx, info.ids); err != nil {
-		log.FromContext(ctx).WithError(err).Warn("Failed to retrieve end device locations")
-	} else {
-		info.uplink.Locations = locations
-	}
 	loc := as.locationFromPayload(info.uplink)
 	if loc != nil {
 		if info.uplink.Locations == nil {
@@ -1285,10 +1298,14 @@ func (as *ApplicationServer) handleSimulatedUplink(ctx context.Context, info upl
 		return err
 	}
 
-	if locations, err := as.locationRegistry.Get(ctx, info.ids); err != nil {
-		log.FromContext(ctx).WithError(err).Warn("Failed to retrieve end device locations")
+	if entity, err := as.endDeviceRegistry.Get(ctx, info.ids, []string{"attributes", "locations"}); err != nil {
+		log.FromContext(ctx).WithError(err).Warn(
+			"Failed to retrieve the end device from entity registry on simulated uplink",
+		)
 	} else {
-		info.uplink.Locations = locations
+		info.uplink.Attributes = entity.Attributes
+		info.uplink.Locations = entity.Locations
+
 	}
 
 	if err := as.decodeUplink(ctx, dev, info.uplink, info.link.DefaultFormatters); err != nil {
@@ -1370,7 +1387,19 @@ func (as *ApplicationServer) handleDownlinkQueueInvalidated(
 			return dev, mask, nil
 		},
 	)
-	return pass, err
+	if err != nil {
+		return pass, err
+	}
+
+	if entity, err := as.endDeviceRegistry.Get(ctx, ids, []string{"attributes"}); err != nil {
+		log.FromContext(ctx).WithError(err).Warn(
+			"Failed to retrieve end device attributes on downlink queue invalidated",
+		)
+	} else {
+		invalid.Attributes = entity.Attributes
+	}
+
+	return pass, nil
 }
 
 func (as *ApplicationServer) handleDownlinkNack(
@@ -1455,17 +1484,47 @@ func (as *ApplicationServer) handleDownlinkNack(
 			return dev, mask, nil
 		},
 	)
-	return err
+	if err != nil {
+		return err
+	}
+
+	if _, err = as.endDeviceRegistry.Set(ctx, ids, []string{"attributes"},
+		func(entity *ttnpb.EndDevice) (*ttnpb.EndDevice, []string, error) {
+			if entity == nil {
+				return nil, nil, errDeviceNotFound.WithAttributes("device_uid", unique.ID(ctx, ids))
+			}
+			msg.Attributes = entity.Attributes
+			return entity, []string{""}, nil
+		},
+	); err != nil {
+		log.FromContext(ctx).WithError(err).Warn(
+			"Failed to retrieve end device attributes on downlink nack",
+		)
+	}
+
+	return nil
 }
 
-// handleLocationSolved saves the provided *ttnpb.ApplicationLocation in the Entity Registry as part of the device locations.
-// Locations provided by other services will be maintained.
+// handleLocationSolved saves the provided *ttnpb.ApplicationLocation in the Entity Registry as part of the device
+// locations. Locations provided by other services will be maintained.
 func (as *ApplicationServer) handleLocationSolved(ctx context.Context, ids *ttnpb.EndDeviceIdentifiers, msg *ttnpb.ApplicationLocation, link *ttnpb.ApplicationLink) error {
 	defer trace.StartRegion(ctx, "handle location solved").End()
 
-	if _, err := as.locationRegistry.Merge(ctx, ids, map[string]*ttnpb.Location{
-		msg.Service: msg.Location,
-	}); err != nil {
+	if _, err := as.endDeviceRegistry.Set(ctx, ids, []string{"locations"},
+		func(stored *ttnpb.EndDevice) (*ttnpb.EndDevice, []string, error) {
+			if stored == nil {
+				return nil, nil, errDeviceNotFound.WithAttributes("device_uid", unique.ID(ctx, ids))
+			}
+
+			if len(stored.Locations) == 0 {
+				stored.Locations = make(map[string]*ttnpb.Location)
+			}
+
+			stored.Locations[msg.Service] = msg.Location
+
+			return stored, []string{"locations"}, nil
+		},
+	); err != nil {
 		log.FromContext(ctx).WithError(err).Warn("Failed to merge end device locations")
 	}
 	return nil
@@ -1486,6 +1545,15 @@ func (as *ApplicationServer) decryptDownlinkMessage(ctx context.Context, ids *tt
 	if err != nil {
 		return err
 	}
+
+	if entity, err := as.endDeviceRegistry.Get(ctx, ids, []string{"attributes"}); err != nil {
+		log.FromContext(ctx).WithError(err).Warn(
+			"Failed to retrieve end device attributes on downlink message",
+		)
+	} else {
+		msg.Attributes = entity.Attributes
+	}
+
 	var session *ttnpb.Session
 	switch {
 	case dev.Session != nil && bytes.Equal(dev.Session.Keys.SessionKeyId, msg.SessionKeyId):

pkg/applicationserver/applicationserver_test.go

@@ -303,9 +303,7 @@ func TestApplicationServer(t *testing.T) {
 			},
 		},
 		EndDeviceMetadataStorage: applicationserver.EndDeviceMetadataStorageConfig{
-			Location: applicationserver.EndDeviceLocationStorageConfig{
-				Registry: metadata.NewNoopEndDeviceLocationRegistry(),
-			},
+			Registry: metadata.NewNoopEndDeviceRegistry(),
 		},
 		Downlinks: applicationserver.DownlinksConfig{
 			ConfirmationConfig: applicationserver.ConfirmationConfig{
@@ -2414,9 +2412,7 @@ func TestSkipPayloadCrypto(t *testing.T) {
 			},
 		},
 		EndDeviceMetadataStorage: applicationserver.EndDeviceMetadataStorageConfig{
-			Location: applicationserver.EndDeviceLocationStorageConfig{
-				Registry: metadata.NewNoopEndDeviceLocationRegistry(),
-			},
+			Registry: metadata.NewNoopEndDeviceRegistry(),
 		},
 		Downlinks: applicationserver.DownlinksConfig{
 			ConfirmationConfig: applicationserver.ConfirmationConfig{
@@ -2916,9 +2912,7 @@ func TestLocationFromPayload(t *testing.T) {
 			},
 		},
 		EndDeviceMetadataStorage: applicationserver.EndDeviceMetadataStorageConfig{
-			Location: applicationserver.EndDeviceLocationStorageConfig{
-				Registry: metadata.NewClusterEndDeviceLocationRegistry(c, (1<<4)*Timeout),
-			},
+			Registry: metadata.NewClusterEndDeviceRegistry(c, (1<<4)*Timeout),
 		},
 		Downlinks: applicationserver.DownlinksConfig{
 			ConfirmationConfig: applicationserver.ConfirmationConfig{
@@ -3101,9 +3095,7 @@ func TestUplinkNormalized(t *testing.T) {
 			},
 		},
 		EndDeviceMetadataStorage: applicationserver.EndDeviceMetadataStorageConfig{
-			Location: applicationserver.EndDeviceLocationStorageConfig{
-				Registry: metadata.NewClusterEndDeviceLocationRegistry(c, (1<<4)*Timeout),
-			},
+			Registry: metadata.NewClusterEndDeviceRegistry(c, (1<<4)*Timeout),
 		},
 		Downlinks: applicationserver.DownlinksConfig{
 			ConfirmationConfig: applicationserver.ConfirmationConfig{