Remove AS uplink storage

CHANGELOG.md

@@ -11,14 +11,19 @@ For details about compatibility between different releases, see the **Commitment
 
 ### Added
 
+- The `as-db purge` command to purge unused data from the Application Server database.
+
 ### Changed
 
 - Instead of retrying application downlinks indefinitely, the Application Server now retries them for a configured number of times. Each `ApplicationDownlink` message contains the `attempt` and `max_attempts` fields to indicate the current and maximum number of attempts for a specific `application downlink`.
 - The Application Server configuration has the `as.downlinks.confirmation.default-retry-attempts` and `as.downlinks.confirmation.max-retry-attempts` fields that configure the allowed number of retries for application downlinks. The default values are `8` for the `as.downlinks.confirmation.default-retry-attempts` and `32` for the `as.downlinks.confirmation.max-retry-attempts`.
 - The `as.downlinks.confirmation.default-retry-attempts` field is used for all application downlinks that were scheduled before this change and for every application downlink that does not have the `max_attempts` field set. On the other hand, the `as.downlinks.confirmation.max-retry-attempts` field ensures that the `max_attempts` field's upper bound is contained and does not exceed its value.
+- The number of historical frames considered for the multi-frame query window size in the LoRa Geolocation Services integration. The window size is now limited between 1 and 16 frames with 16 being the default value.
 
 ### Deprecated
 
+- The `as.uplink-storage.limit` configuration option.
+
 ### Removed
 
 ### Fixed

cmd/internal/shared/applicationserver/config.go

@@ -48,9 +48,6 @@ var DefaultApplicationServerConfig = applicationserver.Config{
 		Workers:   1024,
 		Downlinks: web.DownlinksConfig{PublicAddress: shared.DefaultPublicURL + "/api/v3"},
 	},
-	UplinkStorage: applicationserver.UplinkStorageConfig{
-		Limit: 16,
-	},
 	EndDeviceMetadataStorage: applicationserver.EndDeviceMetadataStorageConfig{
 		Location: applicationserver.EndDeviceLocationStorageConfig{
 			Timeout: 5 * time.Second,

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

@@ -227,6 +227,39 @@ var (
 			return nil
 		},
 	}
+	asDBPurgeCommand = &cobra.Command{
+		Use:   "purge",
+		Short: "Purge unused Application Server data",
+		RunE: func(cmd *cobra.Command, args []string) error {
+			ctx := cmd.Context()
+
+			logger.Info("Connecting to Redis database...")
+			cl := ttnredis.New(config.Redis.WithNamespace("as", "applicationups"))
+
+			logger.Info("Purging uplink registry data")
+
+			purged := 0
+			uidApplicationUpsKey := cl.Key("uid", "*")
+			pipeliner := cl.Pipeline()
+			err := ttnredis.RangeRedisKeys(ctx, cl, uidApplicationUpsKey, ttnredis.DefaultRangeCount,
+				func(k string) (bool, error) {
+					pipeliner.Del(ctx, k)
+					purged++
+					return true, nil
+				})
+			if err != nil {
+				logger.WithError(err).Error("Failed to purge uplink registry data")
+				return err
+			}
+			if _, err := pipeliner.Exec(ctx); err != nil {
+				logger.WithError(err).Error("Failed to purge uplink registry data")
+				return err
+			}
+
+			logger.WithField("records_purged_count", purged).Info("Purged uplink registry data")
+			return nil
+		},
+	}
 )
 
 func init() {
@@ -236,4 +269,5 @@ func init() {
 	asDBCleanupCommand.Flags().Bool("dry-run", false, "Dry run")
 	asDBCleanupCommand.Flags().Duration("pagination-delay", 100, "Delay between batch requests")
 	asDBCommand.AddCommand(asDBCleanupCommand)
+	asDBCommand.AddCommand(asDBPurgeCommand)
 }

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

@@ -51,18 +51,16 @@ func NewPubSubCleaner(ctx context.Context, config *redis.Config) (*pubsub.Regist
 // NewPackagesCleaner returns a new instance of packages RegistryCleaner with a local set
 // of applications and devices.
 func NewPackagesCleaner(ctx context.Context, config *redis.Config) (*packages.RegistryCleaner, error) {
-	applicationPackagesRegistry := &packageredis.ApplicationPackagesRegistry{
-		Redis:   redis.New(config.WithNamespace("as", "io", "applicationpackages")),
-		LockTTL: defaultLockTTL,
-	}
-	if err := applicationPackagesRegistry.Init(ctx); err != nil {
+	applicationPackagesRegistry, err := packageredis.NewApplicationPackagesRegistry(
+		ctx, redis.New(config.WithNamespace("as", "io", "applicationpackages")), defaultLockTTL,
+	)
+	if err != nil {
 		return nil, shared.ErrInitializeApplicationServer.WithCause(err)
 	}
 	cleaner := &packages.RegistryCleaner{
 		ApplicationPackagesRegistry: applicationPackagesRegistry,
 	}
-	err := cleaner.RangeToLocalSet(ctx)
-	if err != nil {
+	if err := cleaner.RangeToLocalSet(ctx); err != nil {
 		return nil, err
 	}
 	return cleaner, nil
@@ -80,9 +78,6 @@ func NewASDeviceRegistryCleaner(ctx context.Context, config *redis.Config) (*as.
 	}
 	cleaner := &as.RegistryCleaner{
 		DevRegistry: deviceRegistry,
-		AppUpsRegistry: &asredis.ApplicationUplinkRegistry{
-			Redis: redis.New(config.WithNamespace("as", "applicationups")),
-		},
 	}
 	err := cleaner.RangeToLocalSet(ctx)
 	if err != nil {

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

@@ -359,10 +359,6 @@ var startCommand = &cobra.Command{
 				return shared.ErrInitializeApplicationServer.WithCause(err)
 			}
 			config.AS.Devices = deviceRegistry
-			config.AS.UplinkStorage.Registry = &asredis.ApplicationUplinkRegistry{
-				Redis: redis.New(config.Redis.WithNamespace("as", "applicationups")),
-				Limit: config.AS.UplinkStorage.Limit,
-			}
 			config.AS.Distribution.Global.PubSub = &asdistribredis.PubSub{
 				Redis: redis.New(config.Cache.Redis.WithNamespace("as", "traffic")),
 			}
@@ -374,11 +370,12 @@ var startCommand = &cobra.Command{
 				return shared.ErrInitializeApplicationServer.WithCause(err)
 			}
 			config.AS.PubSub.Registry = pubsubRegistry
-			applicationPackagesRegistry := &asioapredis.ApplicationPackagesRegistry{
-				Redis:   redis.New(config.Redis.WithNamespace("as", "io", "applicationpackages")),
-				LockTTL: defaultLockTTL,
-			}
-			if err := applicationPackagesRegistry.Init(ctx); err != nil {
+			applicationPackagesRegistry, err := asioapredis.NewApplicationPackagesRegistry(
+				ctx,
+				redis.New(config.Redis.WithNamespace("as", "io", "applicationpackages")),
+				defaultLockTTL,
+			)
+			if err != nil {
 				return shared.ErrInitializeApplicationServer.WithCause(err)
 			}
 			config.AS.Packages.Registry = applicationPackagesRegistry

config/messages.json

@@ -2285,6 +2285,24 @@
       "file": "messages.go"
     }
   },
+  "error:pkg/applicationserver/io/packages/loragls/v3:decoding_field": {
+    "translations": {
+      "en": "decoding field `{field}`"
+    },
+    "description": {
+      "package": "pkg/applicationserver/io/packages/loragls/v3",
+      "file": "data.go"
+    }
+  },
+  "error:pkg/applicationserver/io/packages/loragls/v3:encoding_field": {
+    "translations": {
+      "en": "encoding field `{field}`"
+    },
+    "description": {
+      "package": "pkg/applicationserver/io/packages/loragls/v3",
+      "file": "data.go"
+    }
+  },
   "error:pkg/applicationserver/io/packages/loragls/v3:field_not_found": {
     "translations": {
       "en": "field `{field}` not found"

pkg/applicationserver/applicationserver.go

@@ -75,13 +75,13 @@ type ApplicationServer struct {
 
 	linkRegistry           LinkRegistry
 	deviceRegistry         DeviceRegistry
-	appUpsRegistry         ApplicationUplinkRegistry
 	locationRegistry       metadata.EndDeviceLocationRegistry
 	formatters             messageprocessors.MapPayloadProcessor
 	webhooks               ioweb.Webhooks
 	webhookTemplates       ioweb.TemplateStore
 	pubsub                 *pubsub.PubSub
 	appPackages            packages.Server
+	appPkgRegistry         packages.Registry
 	deviceLastSeenProvider lastseen.LastSeenProvider
 
 	clusterDistributor distribution.Distributor
@@ -153,7 +153,7 @@ func New(c *component.Component, conf *Config) (as *ApplicationServer, err error
 		config:           conf,
 		linkRegistry:     conf.Links,
 		deviceRegistry:   wrapEndDeviceRegistryWithReplacedFields(conf.Devices, replacedEndDeviceFields...),
-		appUpsRegistry:   conf.UplinkStorage.Registry,
+		appPkgRegistry:   conf.Packages.Registry,
 		locationRegistry: conf.EndDeviceMetadataStorage.Location.Registry,
 		formatters:       make(messageprocessors.MapPayloadProcessor),
 		clusterDistributor: distribution.NewPubSubDistributor(
@@ -1072,38 +1072,6 @@ func (as *ApplicationServer) matchSession(ctx context.Context, ids *ttnpb.EndDev
 	return mask, nil
 }
 
-// storeUplink stores the provided *ttnpb.ApplicationUplink in the device uplink storage.
-// Only fields which are used by integrations are stored.
-// The fields which are stored are based on the following usages:
-// - io/packages/loragls/v3/package.go#multiFrameQuery
-// - io/packages/loragls/v3/api/objects.go#parseRxMetadata.
-func (as *ApplicationServer) storeUplink(
-	ctx context.Context,
-	ids *ttnpb.EndDeviceIdentifiers,
-	uplink *ttnpb.ApplicationUplink,
-) error {
-	cleanUplink := &ttnpb.ApplicationUplink{
-		RxMetadata: make([]*ttnpb.RxMetadata, 0, len(uplink.RxMetadata)),
-		ReceivedAt: uplink.ReceivedAt,
-	}
-	for _, md := range uplink.RxMetadata {
-		if md.GatewayIds == nil {
-			continue
-		}
-		cleanUplink.RxMetadata = append(cleanUplink.RxMetadata, &ttnpb.RxMetadata{
-			GatewayIds: &ttnpb.GatewayIdentifiers{
-				GatewayId: md.GatewayIds.GatewayId,
-			},
-			AntennaIndex:  md.AntennaIndex,
-			FineTimestamp: md.FineTimestamp,
-			Location:      md.Location,
-			Rssi:          md.Rssi,
-			Snr:           md.Snr,
-		})
-	}
-	return as.appUpsRegistry.Push(ctx, ids, cleanUplink)
-}
-
 // setActivated attempts to mark the end device as activated in the Entity Registry.
 // If the update succeeds, the end device will be updated in the Application Server end device registry
 // in order to avoid subsequent calls.
@@ -1220,9 +1188,6 @@ func (as *ApplicationServer) handleUplink(ctx context.Context, info uplinkInfo)
 		if err := as.publishNormalizedUplink(ctx, info); err != nil {
 			return err
 		}
-		if err := as.storeUplink(ctx, info.ids, info.uplink); err != nil {
-			return err
-		}
 	} else if appSKey := dev.GetSession().GetKeys().GetAppSKey(); appSKey != nil {
 		info.uplink.AppSKey = appSKey
 		info.uplink.LastAFCntDown = dev.Session.LastAFCntDown
@@ -1511,11 +1476,6 @@ func (as *ApplicationServer) GetMQTTConfig(ctx context.Context) (*config.MQTT, e
 	return &cfg.MQTT, nil
 }
 
-// RangeUplinks ranges the application uplinks and calls the callback function, until false is returned.
-func (as *ApplicationServer) RangeUplinks(ctx context.Context, ids *ttnpb.EndDeviceIdentifiers, paths []string, f func(ctx context.Context, up *ttnpb.ApplicationUplink) bool) error {
-	return as.appUpsRegistry.Range(ctx, ids, paths, f)
-}
-
 // GetEndDevice retrieves the end device associated with the provided identifiers from the end device registry.
 func (as *ApplicationServer) GetEndDevice(ctx context.Context, ids *ttnpb.EndDeviceIdentifiers, paths []string) (*ttnpb.EndDevice, error) {
 	return as.deviceRegistry.Get(ctx, ids, paths)

pkg/applicationserver/applicationserver_test.go

@@ -215,14 +215,6 @@ func TestApplicationServer(t *testing.T) {
 		t.Fatalf("Failed to set link in registry: %s", err)
 	}
 
-	applicationUpsRedisClient, applicationUpsFlush := test.NewRedis(ctx, "applicationserver_test", "applicationups")
-	defer applicationUpsFlush()
-	defer applicationUpsRedisClient.Close()
-	applicationUpsRegistry := &redis.ApplicationUplinkRegistry{
-		Redis: applicationUpsRedisClient,
-		Limit: 16,
-	}
-
 	webhooksRedisClient, webhooksFlush := test.NewRedis(ctx, "applicationserver_test", "webhooks")
 	defer webhooksFlush()
 	defer webhooksRedisClient.Close()
@@ -296,10 +288,6 @@ func TestApplicationServer(t *testing.T) {
 	config := &applicationserver.Config{
 		Devices: deviceRegistry,
 		Links:   linkRegistry,
-		UplinkStorage: applicationserver.UplinkStorageConfig{
-			Registry: applicationUpsRegistry,
-			Limit:    16,
-		},
 		MQTT: config.MQTT{
 			Listen: ":1883",
 		},
@@ -2399,14 +2387,6 @@ func TestSkipPayloadCrypto(t *testing.T) {
 		t.Fatalf("Failed to set link in registry: %s", err)
 	}
 
-	applicationUpsRedisClient, applicationUpsFlush := test.NewRedis(ctx, "applicationserver_test", "applicationups")
-	defer applicationUpsFlush()
-	defer applicationUpsRedisClient.Close()
-	applicationUpsRegistry := &redis.ApplicationUplinkRegistry{
-		Redis: applicationUpsRedisClient,
-		Limit: 16,
-	}
-
 	distribRedisClient, distribFlush := test.NewRedis(ctx, "applicationserver_test", "traffic")
 	defer distribFlush()
 	defer distribRedisClient.Close()
@@ -2434,10 +2414,6 @@ func TestSkipPayloadCrypto(t *testing.T) {
 	config := &applicationserver.Config{
 		Devices: deviceRegistry,
 		Links:   linkRegistry,
-		UplinkStorage: applicationserver.UplinkStorageConfig{
-			Registry: applicationUpsRegistry,
-			Limit:    16,
-		},
 		Distribution: applicationserver.DistributionConfig{
 			Global: applicationserver.GlobalDistributorConfig{
 				PubSub: distribPubSub,
@@ -2926,14 +2902,6 @@ func TestLocationFromPayload(t *testing.T) {
 	defer distribRedisClient.Close()
 	distribPubSub := distribredis.PubSub{Redis: distribRedisClient}
 
-	applicationUpsRedisClient, applicationUpsFlush := test.NewRedis(ctx, "applicationserver_test", "applicationups")
-	defer applicationUpsFlush()
-	defer applicationUpsRedisClient.Close()
-	applicationUpsRegistry := &redis.ApplicationUplinkRegistry{
-		Redis: applicationUpsRedisClient,
-		Limit: 16,
-	}
-
 	c := componenttest.NewComponent(t, &component.Config{
 		ServiceBase: config.ServiceBase{
 			GRPC: config.GRPC{
@@ -2951,10 +2919,6 @@ func TestLocationFromPayload(t *testing.T) {
 	config := &applicationserver.Config{
 		Devices: deviceRegistry,
 		Links:   linkRegistry,
-		UplinkStorage: applicationserver.UplinkStorageConfig{
-			Registry: applicationUpsRegistry,
-			Limit:    16,
-		},
 		Distribution: applicationserver.DistributionConfig{
 			Global: applicationserver.GlobalDistributorConfig{
 				PubSub: distribPubSub,
@@ -3126,14 +3090,6 @@ func TestUplinkNormalized(t *testing.T) {
 	defer distribRedisClient.Close()
 	distribPubSub := distribredis.PubSub{Redis: distribRedisClient}
 
-	applicationUpsRedisClient, applicationUpsFlush := test.NewRedis(ctx, "applicationserver_test", "applicationups")
-	defer applicationUpsFlush()
-	defer applicationUpsRedisClient.Close()
-	applicationUpsRegistry := &redis.ApplicationUplinkRegistry{
-		Redis: applicationUpsRedisClient,
-		Limit: 16,
-	}
-
 	c := componenttest.NewComponent(t, &component.Config{
 		ServiceBase: config.ServiceBase{
 			GRPC: config.GRPC{
@@ -3151,10 +3107,6 @@ func TestUplinkNormalized(t *testing.T) {
 	config := &applicationserver.Config{
 		Devices: deviceRegistry,
 		Links:   linkRegistry,
-		UplinkStorage: applicationserver.UplinkStorageConfig{
-			Registry: applicationUpsRegistry,
-			Limit:    16,
-		},
 		Distribution: applicationserver.DistributionConfig{
 			Global: applicationserver.GlobalDistributorConfig{
 				PubSub: distribPubSub,
@@ -3392,22 +3344,13 @@ func TestApplicationServerCleanup(t *testing.T) {
 		t.FailNow()
 	}
 
-	applicationUpsRedisClient, applicationUpsFlush := test.NewRedis(ctx, "applicationserver_test", "applicationups")
-	defer applicationUpsFlush()
-	defer applicationUpsRedisClient.Close()
-	applicationUpsRegistry := &redis.ApplicationUplinkRegistry{
-		Redis: applicationUpsRedisClient,
-		Limit: 16,
-	}
-
 	applicationPackagesRedisClient, applicationPackagesFlush := test.NewRedis(ctx, "applicationserver_test", "applicationpackages")
 	defer applicationPackagesFlush()
 	defer applicationPackagesRedisClient.Close()
-	applicationPackagesRegistry := &asioapredis.ApplicationPackagesRegistry{
-		Redis:   applicationPackagesRedisClient,
-		LockTTL: test.Delay << 10,
-	}
-	if err := applicationPackagesRegistry.Init(ctx); !a.So(err, should.BeNil) {
+	applicationPackagesRegistry, err := asioapredis.NewApplicationPackagesRegistry(
+		ctx, applicationPackagesRedisClient, test.Delay<<10,
+	)
+	if !a.So(err, should.BeNil) {
 		t.FailNow()
 	}
 
@@ -3538,7 +3481,7 @@ func TestApplicationServerCleanup(t *testing.T) {
 	pubsubCleaner := &pubsub.RegistryCleaner{
 		PubSubRegistry: pubsubRegistry,
 	}
-	err := pubsubCleaner.RangeToLocalSet(ctx)
+	err = pubsubCleaner.RangeToLocalSet(ctx)
 	a.So(err, should.BeNil)
 	a.So(pubsubCleaner.LocalSet, should.HaveLength, 3)
 
@@ -3550,8 +3493,7 @@ func TestApplicationServerCleanup(t *testing.T) {
 	a.So(webhookCleaner.LocalSet, should.HaveLength, 3)
 
 	devCleaner := &applicationserver.RegistryCleaner{
-		DevRegistry:    deviceRegistry,
-		AppUpsRegistry: applicationUpsRegistry,
+		DevRegistry: deviceRegistry,
 	}
 	err = devCleaner.RangeToLocalSet(ctx)
 	a.So(err, should.BeNil)