Merge pull request #1442 from anotherchrisberry/tag-buffer

EntityTags: batch bulk operations in groups of 50
spinnaker · Feb 20, 2017 · be4644c · be4644c
2 parents cf505b5 + 44620fa
commit be4644c
Show file tree

Hide file tree

Showing 2 changed files with 18 additions and 13 deletions.
diff --git a/.../netflix/spinnaker/clouddriver/elasticsearch/ops/BulkUpsertEntityTagsAtomicOperation.java b/.../netflix/spinnaker/clouddriver/elasticsearch/ops/BulkUpsertEntityTagsAtomicOperation.java
@@ -16,6 +16,7 @@
 
 package com.netflix.spinnaker.clouddriver.elasticsearch.ops;
 
+import com.google.common.collect.Lists;
 import com.netflix.spinnaker.clouddriver.core.services.Front50Service;
 import com.netflix.spinnaker.clouddriver.data.task.Task;
 import com.netflix.spinnaker.clouddriver.data.task.TaskRepository;
@@ -62,19 +63,23 @@ public BulkUpsertEntityTagsAtomicOperationResult operate(List priorOutputs) {
     mergeTags(bulkUpsertEntityTagsDescription);
 
     Date now = new Date();
-    getTask().updateStatus(BASE_PHASE, "Retrieving current entity tags");
 
-    Map<String, EntityTags> existingTags = retrieveExistingTags(entityTags);
+    Lists.partition(entityTags, 50).forEach(tags -> {
 
-    getTask().updateStatus(BASE_PHASE, "Merging existing tags and metadata");
+      getTask().updateStatus(BASE_PHASE, "Retrieving current entity tags");
+      Map<String, EntityTags> existingTags = retrieveExistingTags(tags);
 
-    entityTags.forEach(tag -> mergeExistingTagsAndMetadata(now, existingTags.get(tag.getId()), tag, bulkUpsertEntityTagsDescription.isPartial));
+      getTask().updateStatus(BASE_PHASE, "Merging existing tags and metadata");
+      tags.forEach(tag -> mergeExistingTagsAndMetadata(now, existingTags.get(tag.getId()), tag, bulkUpsertEntityTagsDescription.isPartial));
 
-    Map<String, EntityTags> durableTags = front50Service.batchUpdate(new ArrayList<>(entityTags))
+      getTask().updateStatus(BASE_PHASE, "Performing batch update to durable tagging service");
+      Map<String, EntityTags> durableTags = front50Service.batchUpdate(new ArrayList<>(tags))
         .stream().collect(Collectors.toMap(EntityTags::getId, Function.identity()));
 
-    updateMetadataFromDurableTagsAndIndex(entityTags, durableTags, result);
-    result.upserted.addAll(entityTags);
+      getTask().updateStatus(BASE_PHASE, "Pushing tags to Elastic Search");
+      updateMetadataFromDurableTagsAndIndex(tags, durableTags, result);
+      result.upserted.addAll(tags);
+    });
     return result;
   }
 

diff --git a/...ix/spinnaker/clouddriver/elasticsearch/ops/BulkUpsertEntityTagsAtomicOperationSpec.groovy b/...ix/spinnaker/clouddriver/elasticsearch/ops/BulkUpsertEntityTagsAtomicOperationSpec.groovy
@@ -56,19 +56,19 @@ class BulkUpsertEntityTagsAtomicOperationSpec extends Specification {
 
   void "should perform bulk operation"() {
     given:
-    (1..11).each { addTag(it) }
+    (1..1000).each { addTag(it) }
 
     when:
     operation.operate([])
 
     then:
-    11 * accountCredentialsProvider.getAll() >> { return [testCredentials] }
-    1 * front50Service.getAllEntityTagsById(_) >> []
-    1 * front50Service.batchUpdate(_) >> {
+    1000 * accountCredentialsProvider.getAll() >> { return [testCredentials] }
+    20 * front50Service.getAllEntityTagsById(_) >> []
+    20 * front50Service.batchUpdate(_) >> {
       description.entityTags.findResults { new EntityTags(id: it.id, lastModified: 123, lastModifiedBy: "unknown")}
     }
-    1 * entityTagsProvider.bulkIndex(description.entityTags)
-    11 * entityTagsProvider.verifyIndex(_)
+    20 * entityTagsProvider.bulkIndex(_)
+    1000 * entityTagsProvider.verifyIndex(_)
   }
 
   void 'should set id and pattern to default if none supplied'() {