feat(batchUpdate): enhance batch update functionality

front50-api/src/main/java/com/netflix/spinnaker/front50/api/model/pipeline/Pipeline.java

@@ -21,9 +21,11 @@
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
+import lombok.EqualsAndHashCode;
 import lombok.Getter;
 import lombok.Setter;
 
+@EqualsAndHashCode(of = {"id", "name", "application"})
 public class Pipeline implements Timestamped {
 
   public static final String TYPE_TEMPLATED = "templatedPipeline";

front50-sql/src/main/kotlin/com/netflix/spinnaker/front50/model/SqlStorageService.kt

@@ -258,63 +258,58 @@ class SqlStorageService(
   }
 
   override fun <T : Timestamped> storeObjects(objectType: ObjectType, allItems: Collection<T>) {
-    // using a lower `chunkSize` to avoid exceeding default packet size limits.
-    allItems.chunked(100).forEach { items ->
-      try {
-        withPool(poolName) {
-          jooq.transactional(sqlRetryProperties.transactions) { ctx ->
+    withPool(poolName) {
+      jooq.transactional(sqlRetryProperties.transactions) { ctx ->
+        // using a lower `chunkSize` to avoid exceeding default packet size limits.
+        allItems.chunked(100).forEach { items ->
+          try {
+            ctx.batch(
+              items.map { item ->
+                val insertPairs = definitionsByType[objectType]!!.getInsertPairs(
+                  objectMapper, item.id.toLowerCase(), item
+                )
+                val updatePairs = definitionsByType[objectType]!!.getUpdatePairs(insertPairs)
+
+                ctx.insertInto(
+                  table(definitionsByType[objectType]!!.tableName),
+                  *insertPairs.keys.map { field(it) }.toTypedArray()
+                )
+                  .values(insertPairs.values)
+                  .onConflict(field("id", String::class.java))
+                  .doUpdate()
+                  .set(updatePairs.mapKeys { field(it.key) })
+              }
+            ).execute()
+          } catch (e: SQLDialectNotSupportedException) {
+            for (item in items) {
+              storeSingleObject(objectType, item.id.toLowerCase(), item)
+            }
+          }
+
+          if (definitionsByType[objectType]!!.supportsHistory) {
             try {
               ctx.batch(
                 items.map { item ->
-                  val insertPairs = definitionsByType[objectType]!!.getInsertPairs(
-                    objectMapper, item.id.toLowerCase(), item
+                  val historyPairs = definitionsByType[objectType]!!.getHistoryPairs(
+                    objectMapper, clock, item.id.toLowerCase(), item
                   )
-                  val updatePairs = definitionsByType[objectType]!!.getUpdatePairs(insertPairs)
 
-                  ctx.insertInto(
-                    table(definitionsByType[objectType]!!.tableName),
-                    *insertPairs.keys.map { field(it) }.toTypedArray()
-                  )
-                    .values(insertPairs.values)
-                    .onConflict(field("id", String::class.java))
-                    .doUpdate()
-                    .set(updatePairs.mapKeys { field(it.key) })
+                  ctx
+                    .insertInto(
+                      table(definitionsByType[objectType]!!.historyTableName),
+                      *historyPairs.keys.map { field(it) }.toTypedArray()
+                    )
+                    .values(historyPairs.values)
+                    .onDuplicateKeyIgnore()
                 }
               ).execute()
             } catch (e: SQLDialectNotSupportedException) {
               for (item in items) {
-                storeSingleObject(objectType, item.id.toLowerCase(), item)
-              }
-            }
-
-            if (definitionsByType[objectType]!!.supportsHistory) {
-              try {
-                ctx.batch(
-                  items.map { item ->
-                    val historyPairs = definitionsByType[objectType]!!.getHistoryPairs(
-                      objectMapper, clock, item.id.toLowerCase(), item
-                    )
-
-                    ctx
-                      .insertInto(
-                        table(definitionsByType[objectType]!!.historyTableName),
-                        *historyPairs.keys.map { field(it) }.toTypedArray()
-                      )
-                      .values(historyPairs.values)
-                      .onDuplicateKeyIgnore()
-                  }
-                ).execute()
-              } catch (e: SQLDialectNotSupportedException) {
-                for (item in items) {
-                  storeSingleObjectHistory(objectType, item.id.toLowerCase(), item)
-                }
+                storeSingleObjectHistory(objectType, item.id.toLowerCase(), item)
               }
             }
           }
         }
-      } catch (e: Exception) {
-        log.error("Unable to store objects (objectType: {}, objectKeys: {})", objectType, items.map { it.id })
-        throw e
       }
     }
   }

front50-sql/src/test/kotlin/com/netflix/spinnaker/front50/model/SqlStorageServiceTests.kt

@@ -30,6 +30,7 @@ import dev.minutest.junit.JUnit5Minutests
 import dev.minutest.rootContext
 import java.time.Clock
 import org.jooq.SQLDialect
+import org.jooq.exception.DataAccessException
 import org.jooq.impl.DSL
 import org.jooq.impl.DSL.field
 import org.jooq.impl.DSL.table
@@ -195,6 +196,37 @@ internal object SqlStorageServiceTests : JUnit5Minutests {
           }
         }
 
+        test("bulk create pipelines atomically") {
+          // verify that pipelines can be bulk created
+          val pipelines = (1..500).map { idx ->
+            Pipeline().apply {
+              id = "pipeline${idx}"
+              name = "pipeline${idx}"
+              lastModified = 100 + idx.toLong()
+              lastModifiedBy = "test"
+              setApplication("application")
+            }
+          }
+
+          // set lastModifiedBy of one of the pipelines to null in order to force an error
+          // and make sure no pipelines are added since additions are done in a single transaction
+          pipelines[250].lastModifiedBy = null
+          expectThrows<DataAccessException> {
+            sqlStorageService.storeObjects(ObjectType.PIPELINE,pipelines)
+            expectThat(
+              jooq.selectCount().from("pipelines").fetchOne(0, Int::class.java)
+            ).isEqualTo(0)
+          }
+
+          // Reset lastModifiedBy to ensure successful bulk creation
+          pipelines[250].lastModifiedBy = "test"
+          sqlStorageService.storeObjects(ObjectType.PIPELINE,pipelines)
+
+          val storedPipelines = sqlStorageService.loadObjects<Pipeline>(ObjectType.PIPELINE, pipelines.map { it.id });
+          expectThat(storedPipelines.size).isEqualTo(500);
+          expectThat(storedPipelines.map { it.id }).isEqualTo(pipelines.map { it.id })
+        }
+
         var lastModifiedMs : Long = 100
         test("loadObjects basic behavior") {
           val objectKeys = mutableSetOf<String>()

front50-web/src/main/java/com/netflix/spinnaker/front50/config/Front50WebConfig.java

@@ -24,6 +24,7 @@
 import com.netflix.spinnaker.filters.AuthenticatedRequestFilter;
 import com.netflix.spinnaker.front50.ItemDAOHealthIndicator;
 import com.netflix.spinnaker.front50.api.validator.PipelineValidator;
+import com.netflix.spinnaker.front50.config.controllers.PipelineControllerConfig;
 import com.netflix.spinnaker.front50.model.application.ApplicationDAO;
 import com.netflix.spinnaker.front50.model.application.ApplicationPermissionDAO;
 import com.netflix.spinnaker.front50.model.delivery.DeliveryRepository;
@@ -58,7 +59,10 @@
 @EnableFiatAutoConfig
 @EnableScheduling
 @Import({PluginsAutoConfiguration.class})
-@EnableConfigurationProperties(StorageServiceConfigurationProperties.class)
+@EnableConfigurationProperties({
+  StorageServiceConfigurationProperties.class,
+  PipelineControllerConfig.class
+})
 public class Front50WebConfig extends WebMvcConfigurerAdapter {
 
   @Autowired private Registry registry;

front50-web/src/main/java/com/netflix/spinnaker/front50/config/controllers/PipelineControllerConfig.java

@@ -0,0 +1,35 @@
+/*
+ * Copyright 2024 Salesforce, Inc.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package com.netflix.spinnaker.front50.config.controllers;
+
+import lombok.Data;
+import org.springframework.boot.context.properties.ConfigurationProperties;
+
+@Data
+@ConfigurationProperties(prefix = "controller.pipeline")
+public class PipelineControllerConfig {
+
+  /** Holds the configurations to be used for save/update controller mappings */
+  private SavePipelineConfiguration save = new SavePipelineConfiguration();
+
+  @Data
+  public static class SavePipelineConfiguration {
+    /** This controls whether cache should be refreshes while checking for duplicate pipelines */
+    boolean refreshCacheOnDuplicatesCheck = true;
+  }
+}