fix(gcs): Fix startup error when bucketLocation is not specified (#934)

* test(gcs): Add some implementations to FakeStorageRpc

In an upcoming commit, we'll need to also support getting a bucket;
to prepare for that let's update the FakeStorageRpc to:
* Keep track of what bucket objects are in (rather than now where it
assumes they are all in the same bucket)
* Handle creating and getting a bucket

We'll add an additional Map around the blobs data structure that
maps the bucket to a Map of blobs.  (This is a bit of a complex
data structure that in production code might be better expressed
as classes, but for this single fake in a test it's porbably fine.)

Creating a bucket initializes its blobs to an empty map; we know
if a bucket exists by whether it has an entry in the map.

I used the documentation on the interface we're mocking to decide
how to handle a missing bucket; in some cases we just return null
(when getting an object or a bucket) while in others we propagate
the 404.

This required one small change to the existing tests, which is that
we'll need to create the bucket before we use them in the tests.

* test(gcs): Add test to demonstrate NPE on missing bucketLocation

This commit adds a broken test to demostrate the NPE that occurs
on startup when the bucketLocation is not specified.

* fix(gcs): Fix startup error when bucketLocation is not specified

There was a regression in 1.22 where omitting bucketLocation from
one's GCS config now causes an error on startup. Prior to the
rewrite of GcsStorageService, we accepted either null or empty
string as the bucket location; now that GcsStorageService is in
kotlin and does not use a String? for the field, an error occurs
on trying to create the GcsStorageService.

The fix is just to default the string to "" in the config properties.

* style(gcs): Use Bucket.of intead of builder

* test(gcs): Replace nested map with classes

To simplify the nested map, define a Buckets class and a
BucketContents class, with operation names that make it more
clear what is happening.

Also, I realized that before my changes we were appending the bucket
name to the object name before putting it into the map, but now we
don't need to do that anymore because we have a separate map for
each bucket, so remove some unecessary calls to fullPath (which is
now only used in printing error messages).

* style(gcs): Fix typo

* style(gcs): Fix constructor

There's no need to accept the map in a private constructor then
separately define a no-arg constructor that passes an empty map,
just make the map a val in the class.

front50-gcs/src/main/java/com/netflix/spinnaker/front50/config/GcsProperties.java

@@ -22,7 +22,7 @@
 public class GcsProperties {
   private String bucket;
 
-  private String bucketLocation;
+  private String bucketLocation = "";
 
   private String rootFolder = "front50";
 

front50-gcs/src/test/kotlin/com/netflix/spinnaker/front50/model/FakeStorageRpc.kt

@@ -53,15 +53,46 @@ internal class FakeStorageRpc(private val clock: Clock) : StorageRpc {
 
   companion object {
     private val ALLOWED_LIST_OPTIONS = setOf(Option.PREFIX, Option.VERSIONS)
+    private fun fullPath(storageObject: StorageObject): String = fullPath(storageObject.bucket, storageObject.name)
+    private fun fullPath(bucket: String, name: String) = "$bucket/$name"
+  }
+
+  /**
+   * Represents a collection of buckets.
+   */
+  private class Buckets {
+    private val buckets: MutableMap<String, BucketContents> = mutableMapOf()
+
+    operator fun get(bucket: String) = buckets[bucket]
+
+    fun exists(bucket: String) = buckets.containsKey(bucket)
+
+    fun create(bucket: String) {
+      buckets[bucket] = BucketContents()
+    }
+  }
+
+  /**
+   * Represents the contents of a bucket.
+   */
+  private class BucketContents {
+    private val objects: MutableMap<String, MutableList<Blob>> = mutableMapOf()
+
+    fun getGenerations(storageObject: StorageObject) = objects.getOrPut(storageObject.name, { mutableListOf() })
+
+    fun delete(storageObject: StorageObject) = objects.remove(storageObject.name) != null
+
+    fun list(prefix: String) = objects.filter { (path, _) -> path.startsWith(prefix) }.values
   }
 
   private data class Blob(val storageObject: StorageObject, val content: ByteArray)
 
-  private val blobs = mutableMapOf<String, MutableList<Blob>>()
+  private val buckets = Buckets()
 
   override fun create(storageObject: StorageObject, data: InputStream, options: MutableMap<Option, *>): StorageObject {
     if (options.isNotEmpty()) throw UnsupportedOperationException("unsupported options to create: ${options.keys}")
     if (storageObject.generation != null) throw UnsupportedOperationException("can't call create with a specific generation")
+    val blobs = buckets[storageObject.bucket] ?: throw StorageException(404, "bucket ${storageObject.bucket} does not exist")
     val generations = blobs.getGenerations(storageObject)
     val stampedObject = storageObject.clone()
     stampedObject.generation = generations.size + 1L
@@ -73,6 +104,7 @@ internal class FakeStorageRpc(private val clock: Clock) : StorageRpc {
   override fun patch(storageObject: StorageObject, options: MutableMap<Option, *>): StorageObject {
     if (options.isNotEmpty()) throw UnsupportedOperationException("unsupported options to patch: ${options.keys}")
     if (storageObject.generation != null) throw UnsupportedOperationException("can't call patch with a specific generation")
+    val blobs = buckets[storageObject.bucket] ?: throw StorageException(404, "bucket ${storageObject.bucket} does not exist")
     val generations = blobs.getGenerations(storageObject)
     if (generations.isEmpty()) throw StorageException(404, "no object ${fullPath(storageObject)}")
     val foundObject = generations[generations.size - 1].storageObject
@@ -83,6 +115,7 @@ internal class FakeStorageRpc(private val clock: Clock) : StorageRpc {
 
   override fun get(storageObject: StorageObject, options: MutableMap<Option, *>): StorageObject? {
     if (options.isNotEmpty()) throw UnsupportedOperationException("unsupported options to get: ${options.keys}")
+    val blobs = buckets[storageObject.bucket] ?: return null
     val generations = blobs.getGenerations(storageObject)
     val generation = storageObject.generation ?: generations.size.toLong()
     if (generation == 0L || generation > generations.size) return null
@@ -91,6 +124,7 @@ internal class FakeStorageRpc(private val clock: Clock) : StorageRpc {
 
   override fun load(storageObject: StorageObject, options: MutableMap<Option, *>): ByteArray? {
     if (options.isNotEmpty()) throw UnsupportedOperationException("unsupported options to load: ${options.keys}")
+    val blobs = buckets[storageObject.bucket] ?: throw StorageException(404, "bucket ${storageObject.bucket} does not exist")
     val generations = blobs.getGenerations(storageObject)
     val generation = storageObject.generation ?: generations.size.toLong()
     if (generation == 0L || generation > generations.size) return null
@@ -100,33 +134,28 @@ internal class FakeStorageRpc(private val clock: Clock) : StorageRpc {
   override fun delete(storageObject: StorageObject, options: MutableMap<Option, *>): Boolean {
     if (options.isNotEmpty()) throw UnsupportedOperationException("unsupported options to delete: ${options.keys}")
     if (storageObject.generation != null) throw UnsupportedOperationException("can't delete generations")
-    return blobs.remove(fullPath(storageObject)) != null
+    val blobs = buckets[storageObject.bucket] ?: return false
+    return blobs.delete(storageObject)
   }
 
   override fun list(bucket: String, options: MutableMap<Option, *>): Tuple<String?, Iterable<StorageObject>> {
     val unsupportedOptions = options - ALLOWED_LIST_OPTIONS
     if (unsupportedOptions.isNotEmpty()) {
       throw java.lang.UnsupportedOperationException("unsupported options to list: $unsupportedOptions")
     }
-    val prefix = if (options.containsKey(Option.PREFIX)) fullPath(bucket, options.get(Option.PREFIX)!! as String) else "$bucket/"
+    val prefix = options[Option.PREFIX] as String? ?: ""
     var versionFilter: (List<Blob>) -> List<Blob> = { listOf(it.last()) }
     if (options.get(Option.VERSIONS) as Boolean? == true) {
       // We want to return them in low- to high-generation order, so this will do it
       versionFilter = { it }
     }
+    val blobs = buckets[bucket] ?: throw StorageException(404, "bucket $bucket does not exist")
     return Tuple.of(
       /* pageToken= */ null,
-      blobs.filter { (path, _) -> path.startsWith(prefix) }.values.flatMap(versionFilter).map { it.storageObject }
+      blobs.list(prefix).flatMap(versionFilter).map { it.storageObject }
     )
   }
 
-  private fun <V> MutableMap<String, MutableList<V>>.getGenerations(storageObject: StorageObject) =
-    getOrPut(fullPath(storageObject), { mutableListOf() })
-
-  private fun fullPath(storageObject: StorageObject): String = fullPath(storageObject.bucket, storageObject.name)
-
-  private fun fullPath(bucket: String, name: String) = "$bucket/$name"
-
   override fun open(storageObject: StorageObject, options: MutableMap<Option, *>): String {
     TODO("Not yet implemented")
   }
@@ -168,7 +197,10 @@ internal class FakeStorageRpc(private val clock: Clock) : StorageRpc {
   }
 
   override fun create(bucket: Bucket?, options: MutableMap<Option, *>?): Bucket {
-    TODO("Not yet implemented")
+    val name = bucket?.name ?: throw java.lang.UnsupportedOperationException("bucket name must be specified")
+    if (buckets.exists(name)) throw StorageException(409, "bucket $bucket already exists")
+    buckets.create(name)
+    return Bucket().setName(name)
   }
 
   override fun updateHmacKey(hmacKeyMetadata: HmacKeyMetadata?, options: MutableMap<Option, *>?): HmacKeyMetadata {
@@ -199,8 +231,9 @@ internal class FakeStorageRpc(private val clock: Clock) : StorageRpc {
     TODO("Not yet implemented")
   }
 
-  override fun get(bucket: Bucket?, options: MutableMap<Option, *>?): Bucket {
-    TODO("Not yet implemented")
+  override fun get(bucket: Bucket?, options: MutableMap<Option, *>?): Bucket? {
+    val name = bucket?.name ?: throw java.lang.UnsupportedOperationException("bucket name must be specified")
+    return if (buckets.exists(name)) Bucket().setName(name) else null
   }
 
   override fun testIamPermissions(bucket: String?, permissions: MutableList<String>?, options: MutableMap<Option, *>?): TestIamPermissionsResponse {

front50-gcs/src/test/kotlin/com/netflix/spinnaker/front50/model/GcsIntegrationTest.kt

@@ -0,0 +1,49 @@
+/*
+ * Copyright 2020 Google, LLC
+ *
+ * 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.spinnnaker.front50.model
+
+import com.netflix.spinnaker.front50.config.GcsConfig
+import com.netflix.spinnaker.front50.model.GcsIntegrationTestConfiguration
+import com.netflix.spinnaker.front50.model.GcsStorageService
+import com.netflix.spinnaker.front50.model.ObjectType
+import com.netflix.spinnaker.front50.model.pipeline.Pipeline
+import org.junit.jupiter.api.Test
+import org.junit.jupiter.api.extension.ExtendWith
+import org.springframework.beans.factory.annotation.Autowired
+import org.springframework.boot.test.context.ConfigFileApplicationContextInitializer
+import org.springframework.test.context.ContextConfiguration
+import org.springframework.test.context.TestPropertySource
+import org.springframework.test.context.junit.jupiter.SpringExtension
+import strikt.api.expectThat
+import strikt.assertions.hasSize
+import strikt.assertions.isEmpty
+
+@ExtendWith(SpringExtension::class)
+@ContextConfiguration(
+  classes = [GcsConfig::class, GcsIntegrationTestConfiguration::class],
+  initializers = [ConfigFileApplicationContextInitializer::class]
+)
+@TestPropertySource(properties = ["spring.config.location=classpath:minimal-gcs-account.yml"])
+class GcsIntegrationTest {
+  @Test
+  fun startupTest(@Autowired storageService: GcsStorageService) {
+    expectThat(storageService.listObjectKeys(ObjectType.PIPELINE)).isEmpty()
+    storageService.storeObject(ObjectType.PIPELINE, "my-key", Pipeline())
+    expectThat(storageService.listObjectKeys(ObjectType.PIPELINE)).hasSize(1)
+  }
+}

front50-gcs/src/test/kotlin/com/netflix/spinnaker/front50/model/GcsIntegrationTestConfiguration.kt

@@ -0,0 +1,58 @@
+/*
+ * Copyright 2020 Google, LLC
+ *
+ * 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.model
+
+import com.google.auth.Credentials
+import com.google.cloud.storage.Storage
+import com.google.cloud.storage.StorageOptions
+import com.netflix.spectator.api.NoopRegistry
+import com.netflix.spectator.api.Registry
+import com.netflix.spinnaker.front50.config.GcsProperties
+import com.netflix.spinnaker.front50.config.StorageServiceConfigurationProperties
+import com.netflix.spinnnaker.front50.model.FakeStorageRpcFactory
+import io.github.resilience4j.circuitbreaker.CircuitBreakerRegistry
+import io.github.resilience4j.circuitbreaker.internal.InMemoryCircuitBreakerRegistry
+import io.mockk.mockk
+import java.time.Clock
+import java.time.Instant
+import java.time.ZoneOffset
+import org.springframework.beans.factory.annotation.Qualifier
+import org.springframework.boot.context.properties.EnableConfigurationProperties
+import org.springframework.boot.test.context.TestConfiguration
+import org.springframework.context.annotation.Bean
+import org.springframework.context.annotation.Primary
+
+@EnableConfigurationProperties(StorageServiceConfigurationProperties::class)
+@TestConfiguration
+class GcsIntegrationTestConfiguration {
+  private val clock = Clock.fixed(Instant.ofEpochSecond(629528400L), ZoneOffset.UTC)
+
+  @Bean
+  fun noopRegistry(): Registry = NoopRegistry()
+
+  @Bean
+  @Primary
+  @Qualifier("gcsCredentials") fun gcsCredentials(): Credentials = mockk()
+
+  @Bean
+  fun circuitBreakerRegistry(): CircuitBreakerRegistry = InMemoryCircuitBreakerRegistry()
+
+  @Bean
+  fun storage(properties: GcsProperties): Storage {
+    return StorageOptions.newBuilder().setServiceRpcFactory(FakeStorageRpcFactory(clock)).build().service
+  }
+}

front50-gcs/src/test/kotlin/com/netflix/spinnaker/front50/model/TestGcsStorageService.kt

@@ -84,7 +84,11 @@ class GcsStorageServiceTest {
     clock = SettableClock()
     gcs = when {
       testInfo.tags.contains("mockGcs") -> mockk()
-      else -> StorageOptions.newBuilder().setServiceRpcFactory(FakeStorageRpcFactory(clock)).build().service
+      else -> {
+        val service = StorageOptions.newBuilder().setServiceRpcFactory(FakeStorageRpcFactory(clock)).build().service
+        service.create(Bucket.of(BUCKET_NAME))
+        service
+      }
     }
     executor = when {
       testInfo.tags.contains("testExecutor") -> ControlledExecutor()

front50-gcs/src/test/resources/minimal-gcs-account.yml

@@ -0,0 +1,5 @@
+spinnaker:
+  gcs:
+    enabled: true
+    project: my-project
+    bucket: my-bucket