Merge pull request #46 from gatewayd-io/replace-keys-with-scan

Replace keys with scan

Author: mostafa

plugin/plugin.go

@@ -268,16 +268,18 @@ func (p *Plugin) invalidateDML(ctx context.Context, query string) {
 		// TODO: This is not efficient. We should be able to invalidate the cache
 		// for a specific key instead of invalidating the entire table.
 		pipeline := p.RedisClient.Pipeline()
+		var cursor uint64
 		for {
-			scanResult := p.RedisClient.Scan(ctx, 0, table+":*", p.ScanCount)
+			scanResult := p.RedisClient.Scan(ctx, cursor, table+":*", p.ScanCount)
 			if scanResult.Err() != nil {
 				CacheMissesCounter.Inc()
 				p.Logger.Debug("Failed to scan keys", "error", scanResult.Err())
 				break
 			}
 
 			// Per each key, delete the cache entry and the table cache key itself.
-			keys, cursor := scanResult.Val()
+			var keys []string
+			keys, cursor = scanResult.Val()
 			for _, tableKey := range keys {
 				// Invalidate the cache for the table.
 				cachedRespnseKey := strings.TrimPrefix(tableKey, table+":")

plugin/scheduler.go

@@ -21,43 +21,57 @@ func (p *Plugin) PeriodicInvalidator() {
 		p.Logger.Trace("Got proxies from GatewayD", "proxies", proxies)
 
 		// Get all the client keys and delete the ones that are not valid.
-		// TODO: use scan instead of keys.
-		for _, address := range p.RedisClient.Keys(context.Background(), "*:*").Val() {
-			valid := false
-
-			// Validate the address if the address is an IP address.
-			if ok, err := validateAddressPort(address); ok && err == nil {
-				valid = true
-			} else {
-				p.Logger.Trace(
-					"Skipping connection because it is invalid", "address", address, "error", err)
+		var cursor uint64
+		for {
+			scanResult := p.RedisClient.Scan(context.Background(), cursor, "*:*", p.ScanCount)
+			if scanResult.Err() != nil {
+				p.Logger.Error("Failed to scan keys", "error", scanResult.Err())
+				break
 			}
 
-			if !valid {
-				// Validate the address if the address is a hostname.
-				if ok, err := validateHostPort(address); ok && err == nil {
+			var addresses []string
+			addresses, cursor = scanResult.Val()
+			for _, address := range addresses {
+				valid := false
+
+				// Validate the address if the address is an IP address.
+				if ok, err := validateAddressPort(address); ok && err == nil {
 					valid = true
 				} else {
 					p.Logger.Trace(
 						"Skipping connection because it is invalid", "address", address, "error", err)
+				}
+
+				if !valid {
+					// Validate the address if the address is a hostname.
+					if ok, err := validateHostPort(address); ok && err == nil {
+						valid = true
+					} else {
+						p.Logger.Trace(
+							"Skipping connection because it is invalid", "address", address, "error", err)
+						continue
+					}
+				}
+
+				// If the address is not valid, skip it.
+				if !valid {
 					continue
 				}
-			}
 
-			// If the address is not valid, skip it.
-			if !valid {
-				continue
-			}
+				// If the connection is busy (a client is connected), it is not safe to delete the key.
+				if isBusy(proxies, address) {
+					p.Logger.Trace("Skipping connection because it is busy", "address", address)
+					continue
+				}
 
-			// If the connection is busy (a client is connected), it is not safe to delete the key.
-			if isBusy(proxies, address) {
-				p.Logger.Trace("Skipping connection because it is busy", "address", address)
-				continue
+				p.RedisClient.Del(context.Background(), address)
+				p.Logger.Trace("Deleted stale address", "address", address)
+				CacheDeletesCounter.Inc()
 			}
 
-			p.RedisClient.Del(context.Background(), address)
-			p.Logger.Trace("Deleted stale address", "address", address)
-			CacheDeletesCounter.Inc()
+			if cursor == 0 {
+				break
+			}
 		}
 	}); err != nil {
 		p.Logger.Error("Failed to start periodic invalidator",