chore(sql): normalize permission schema (#849)

* chore(sql): normalize permission schema

Normalizes the permissions schema to improve scalability and optimize
code based on the new schema.

The biggest change is that resources are now stored once rather than
duplicated for each user. This means bulk operations are much faster at
the cost of more queries.

* chore(sql): batch up SQL queries to cut round trip time

* chore(sql): add index to updated_at column

This speeds up the delete stage of bulk updates.

* chore(sql): select distinct is cheaper on getAllByRoles

No cost change for `getAllById`

* chore(sql): actually drop the unneeded group by

* chore(sql): optimise for update case

* chore(sql): unify update code between put and putAllById

* chore(sql): reduce permission update to one statement

* chore(sql): more efficient delete and split put tx

* chore(sql): fixes handling of put during sync

Fixes issues where resources added during a sync would break the sync
processes attempt to tidy up unused resources.

Also speeds up deletes for unused permissions for users.

* perf(sql): remove updated_at

The updated_at column isn't needed on permissions and resources now. So
remove it as it's expensive to update.

* chore(sql): make put operation transactional

* chore(sql): don't issue deletes we don't need to

* chore(sql): refactor table definitions

Split up the table definitions to be easier to read and manage. Layout
is based on what jooq would generate, but I don't really want to introduce
code generation for three tables.

* chore(sql): fix package location of tests

* feat(sql): handle extension resources as well as the built-in ones

* fix(sql): don't delete unrestricted user on sync

* chore(sql): fix copyright header.

* fix(sql): fix concurrent puts on users and resources

* fix(sql): fix updating resource timestamps

* chore(sql): update resources only when needed

Refactor the code managing updates to work more like clouddriver and
only insert when resources have actually changed. Should dramatically
drop the number of writes and thus deadlocks.

Adds a unit test as well that attempts to verify the lack of deadlocks.

* chore(sql): add missing column definition

* chore(idea): undo changes to idea settings

* chore(idea): really revert unintended config change

* perf(sql): faster query by role

Use a semi-join to speed up querying permissions by role.

* perf(sql): use semi-join to speed up get of user

* bug(sql): fix reading of resources for a user

* bug(sql): fix integration tests when using sql

* fix(sql): don't delete users in putAllById

The set of users passed in doesn't always have the complete list
for various reasons so it can lead to inconsistency in the data store.

This also matches the behaviour of the Redis storage backend.

* perf(sql): faster queries in getAllByRoles

* fix(sql): don't self merge unrestricted user permissions

* perf(sql): parse resources and permissions in parallel

* fix(sql): typo in property name

* fix(sql): borrow async code from clouddriver

* feat(sql): concurrent writes during putAllById

* perf(sql): rewrite around clouddriver style batching

* chore(sql): fix formatting

* perf(sql): use same async config behaviour as clouddriver

* perf(sql): cheaper query for user permissions

* perf(sql): split up user and permission reading

* perf(sql): split out user write

* perf(sql): revert user read/write splitting

Net performance regression in testing.

* perf(sql): split up permission set read

Co-authored-by: mergify[bot] <37929162+mergify[bot]@users.noreply.github.com>

fiat-core/src/main/java/com/netflix/spinnaker/fiat/model/resources/ResourceType.java

@@ -35,6 +35,10 @@ public ResourceType(String name) {
     this.hashCode = Objects.hash(name);
   }
 
+  public String getName() {
+    return name;
+  }
+
   // TODO(ttomsu): This is Redis-specific, so it probably shouldn't go here.
   public static ResourceType parse(String pluralOrKey) {
     if (pluralOrKey == null) {

fiat-sql/fiat-sql.gradle

@@ -34,6 +34,7 @@ dependencies {
     implementation "io.github.resilience4j:resilience4j-retry"
 
     implementation "org.jetbrains.kotlinx:kotlinx-coroutines-core"
+    implementation "org.jetbrains.kotlinx:kotlinx-coroutines-slf4j"
 
     testImplementation "io.mockk:mockk"
     testImplementation "dev.minutest:minutest"

fiat-sql/src/main/kotlin/com/netflix/spinnaker/config/SqlConfiguration.kt

@@ -20,40 +20,86 @@ import com.netflix.spectator.api.Registry
 import com.netflix.spinnaker.fiat.model.resources.Resource
 import com.netflix.spinnaker.fiat.permissions.PermissionsRepository
 import com.netflix.spinnaker.fiat.permissions.SqlPermissionsRepository
+import com.netflix.spinnaker.kork.dynamicconfig.DynamicConfigService
 import com.netflix.spinnaker.kork.sql.config.DefaultSqlConfiguration
 import com.netflix.spinnaker.kork.sql.config.SqlProperties
 import com.netflix.spinnaker.kork.telemetry.InstrumentedProxy
+import kotlinx.coroutines.ObsoleteCoroutinesApi
+import kotlinx.coroutines.newFixedThreadPoolContext
+import kotlinx.coroutines.slf4j.MDCContext
 import org.jooq.DSLContext
+import org.slf4j.LoggerFactory
+import org.springframework.beans.factory.annotation.Value
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty
 import org.springframework.context.annotation.Bean
 import org.springframework.context.annotation.Configuration
 import org.springframework.context.annotation.Import
 import java.time.Clock
+import kotlin.contracts.ExperimentalContracts
 
+const val coroutineThreadPrefix = "sql"
+
+@ExperimentalContracts
 @Configuration
 @ConditionalOnProperty("sql.enabled")
 @Import(DefaultSqlConfiguration::class)
 class SqlConfiguration {
 
+    companion object {
+        private val log = LoggerFactory.getLogger(SqlConfiguration::class.java)
+    }
+
+    /*
+     * permissionsRepository.sql.asyncPoolSize: If set to a positive integer, a fixed thread pool of this size is created
+     * as part of a coroutineContext. If permissionsRepository.sql.maxQueryConcurrency is also >1 (default value: 4),
+     * sql queries to fetch > 2 * permissionsRepository.sql.readBatchSize values will be made asynchronously in batches of
+     * maxQueryConcurrency size.
+     */
     @ConditionalOnProperty("permissions-repository.sql.enabled")
     @Bean
+    @ObsoleteCoroutinesApi
     fun sqlPermissionsRepository(
         objectMapper: ObjectMapper,
         registry: Registry,
         jooq: DSLContext,
         sqlProperties: SqlProperties,
-        resources: List<Resource>
-    ): PermissionsRepository =
-        SqlPermissionsRepository(
+        resources: List<Resource>,
+        dynamicConfigService: DynamicConfigService,
+        @Value("\${permissions-repository.sql.async-pool-size:0}") poolSize: Int
+    ): PermissionsRepository {
+
+        /**
+         * newFixedThreadPoolContext was marked obsolete in Oct 2018, to be reimplemented as a new
+         * concurrency limiting threaded context factory with reduced context switch overhead. As of
+         * Feb 2019, the new implementation is unreleased. See: https://github.com/Kotlin/kotlinx.coroutines/issues/261
+         *
+         * TODO: switch to newFixedThreadPoolContext's replacement when ready
+         */
+        val dispatcher = if (poolSize < 1) {
+            null
+        } else {
+            newFixedThreadPoolContext(nThreads = poolSize, name = coroutineThreadPrefix) + MDCContext()
+        }
+
+        if (dispatcher != null) {
+            log.info("Configured coroutine context with newFixedThreadPoolContext of $poolSize threads")
+        }
+
+        return SqlPermissionsRepository(
             Clock.systemUTC(),
             objectMapper,
             jooq,
             sqlProperties.retries,
-            if (sqlProperties.connectionPools.keys.size > 1)
-                sqlProperties.connectionPools.filter { it.value.default }.keys.first() else sqlProperties.connectionPools.keys.first()
-            ,
-            resources
+            resources,
+            dispatcher,
+            dynamicConfigService
         ).let {
-            InstrumentedProxy.proxy(registry, it, "permissionsRepository", mapOf(Pair("repositoryType", "sql"))) as PermissionsRepository
+            InstrumentedProxy.proxy(
+                registry,
+                it,
+                "permissionsRepository",
+                mapOf(Pair("repositoryType", "sql"))
+            ) as PermissionsRepository
         }
+    }
 }