fix(redis): hydrating all values not just first 100 (#152)

spinnaker · Nov 13, 2018 · 0a90e17 · 0a90e17
1 parent f88120b
commit 0a90e17
Show file tree

Hide file tree

Showing 7 changed files with 52 additions and 16 deletions.
diff --git a/swabbie-core/src/main/kotlin/com/netflix/spinnaker/swabbie/AbstractResourceTypeHandler.kt b/swabbie-core/src/main/kotlin/com/netflix/spinnaker/swabbie/AbstractResourceTypeHandler.kt
@@ -667,6 +667,8 @@ abstract class AbstractResourceTypeHandler<T : Resource>(
         return
       }
 
+      log.info("Processing ${markedResources.size} for notification in ${javaClass.simpleName}")
+
       val maxItemsToProcess = Math.min(markedResources.size, workConfiguration.maxItemsProcessedPerCycle)
       markedResources.subList(0, maxItemsToProcess)
         .filter {

diff --git a/swabbie-core/src/main/kotlin/com/netflix/spinnaker/swabbie/CacheStatus.kt b/swabbie-core/src/main/kotlin/com/netflix/spinnaker/swabbie/CacheStatus.kt
@@ -31,7 +31,7 @@ open class InMemoryCacheStatus(
           shutdown()
         }
       } catch (e: Exception) {
-        log.error("Failed while checking the caches in ${javaClass.simpleName}.")
+        log.error("Failed while checking the caches in ${javaClass.simpleName}.", e)
       }
     }, 0, 5, TimeUnit.SECONDS)
   }

diff --git a/swabbie-core/src/main/kotlin/com/netflix/spinnaker/swabbie/agents/ScheduledAgent.kt b/swabbie-core/src/main/kotlin/com/netflix/spinnaker/swabbie/agents/ScheduledAgent.kt
@@ -79,7 +79,7 @@ abstract class ScheduledAgent(
          startupExecutorService.shutdown()
        }
      } catch (e: Exception) {
-       log.error("Failed while waiting for cache to start in ${javaClass.simpleName}.")
+       log.error("Failed while waiting for cache to start in ${javaClass.simpleName}.", e)
      }
     }, 0, 5, TimeUnit.SECONDS)
   }

diff --git a/...redis/src/main/kotlin/com/netflix/spinnaker/swabbie/redis/RedisResourceStateRepository.kt b/...redis/src/main/kotlin/com/netflix/spinnaker/swabbie/redis/RedisResourceStateRepository.kt
@@ -73,13 +73,16 @@ class RedisResourceStateRepository(
         if (set.isEmpty()) {
           emptyList()
         } else {
-          set.chunked(REDIS_CHUNK_SIZE).map { sublist ->
-            this.withCommandsClient<Set<String>> { client ->
-              client.hmget(SINGLE_STATE_KEY, *sublist.map { it }.toTypedArray()).toSet()
+          val state = mutableSetOf<ResourceState>()
+          set.chunked(REDIS_CHUNK_SIZE).forEach { subset ->
+            val partialState = this.withCommandsClient<Set<String>> { client ->
+              client.hmget(SINGLE_STATE_KEY, *subset.map { it }.toTypedArray()).toSet()
             }.map { json ->
               objectMapper.readValue<ResourceState>(json)
             }
-          }.flatten()
+            state.addAll(partialState)
+          }
+          state.toList()
         }
       }
   }

diff --git a/...is/src/main/kotlin/com/netflix/spinnaker/swabbie/redis/RedisResourceTrackingRepository.kt b/...is/src/main/kotlin/com/netflix/spinnaker/swabbie/redis/RedisResourceTrackingRepository.kt
@@ -125,13 +125,17 @@ class RedisResourceTrackingRepository(
 
   private fun hydrateMarkedResources(resourseIds: Set<String>): List<MarkedResource> {
     if (resourseIds.isEmpty()) return emptyList()
-    return resourseIds.chunked(REDIS_CHUNK_SIZE).map { sublist ->
-      return redisClientDelegate.withCommandsClient<Set<String>> { client ->
-          client.hmget(SINGLE_RESOURCES_KEY, *sublist.toTypedArray()).toSet()
-        }.map { json ->
-          objectMapper.readValue<MarkedResource>(json)
-        }
-    }.flatten()
+
+    val hydratedResources = mutableListOf<MarkedResource>()
+    resourseIds.chunked(REDIS_CHUNK_SIZE).forEach { sublist ->
+      val hydrated = redisClientDelegate.withCommandsClient<Set<String>> { client ->
+        client.hmget(SINGLE_RESOURCES_KEY, *sublist.toTypedArray()).toSet()
+      }.map { json ->
+        objectMapper.readValue<MarkedResource>(json)
+      }
+      hydratedResources.addAll(hydrated)
+    }
+    return hydratedResources
   }
 
   override fun upsert(markedResource: MarkedResource, deleteScore: Long, softDeleteScore: Long) {

diff --git a/...src/main/kotlin/com/netflix/spinnaker/swabbie/redis/RedisResourceUseTrackingRepository.kt b/...src/main/kotlin/com/netflix/spinnaker/swabbie/redis/RedisResourceUseTrackingRepository.kt
@@ -126,13 +126,17 @@ class RedisResourceUseTrackingRepository(
 
   private fun hydrateLastSeen(keys: Set<String>): List<LastSeenInfo> {
     if (keys.isEmpty()) return emptyList()
-    return keys.chunked(REDIS_CHUNK_SIZE).map { sublist ->
-      redisClientDelegate.withCommandsClient<Set<String>> { client ->
+
+    val hydratedLastSeen = mutableListOf<LastSeenInfo>()
+    keys.chunked(REDIS_CHUNK_SIZE).forEach { sublist ->
+      val hydrated = redisClientDelegate.withCommandsClient<Set<String>> { client ->
         client.hmget(LAST_SEEN, *sublist.toTypedArray()).toSet()
       }.map { json ->
         objectMapper.readValue<LastSeenInfo>(json)
       }
-    }.flatten()
+      hydratedLastSeen.addAll(hydrated)
+    }
+    return hydratedLastSeen
   }
 
   override fun isUnused(resourceIdentifier: String): Boolean {

diff --git a/...rc/test/kotlin/com/netflix/spinnaker/swabbie/redis/RedisResourceTrackingRepositoryTest.kt b/...rc/test/kotlin/com/netflix/spinnaker/swabbie/redis/RedisResourceTrackingRepositoryTest.kt
@@ -56,6 +56,19 @@ object RedisResourceTrackingRepositoryTest {
     RedisClientSelector(listOf(JedisClientDelegate("primaryDefault", jedisPool))), objectMapper, clock
   )
 
+  private val defaultMarkedResource = MarkedResource(
+    resource = TestResource(resourceId = "test"),
+    summaries = listOf(Summary("invalid resourceHash 1", "rule 1")),
+    namespace = "namespace",
+    projectedDeletionStamp = 1,
+    projectedSoftDeletionStamp = 1,
+    notificationInfo = NotificationInfo(
+      recipient = "yolo@netflixcom",
+      notificationType = "Email",
+      notificationStamp = clock.instant().toEpochMilli()
+    )
+  )
+
   @BeforeEach
   fun setup() {
     jedisPool.resource.use {
@@ -216,4 +229,14 @@ object RedisResourceTrackingRepositoryTest {
 
     resourceRepository.getNumMarkedResources() shouldMatch equalTo(1L)
   }
+
+  @Test
+  fun `scanning should work`() {
+    for(i in 1..200) {
+      resourceRepository.upsert(defaultMarkedResource.copy(resource = TestResource(resourceId = "$i")))
+    }
+
+    resourceRepository.getNumMarkedResources() shouldMatch equalTo(200L)
+    resourceRepository.getMarkedResources().size shouldMatch equalTo(200)
+  }
 }