chore: ffs: feature flag service (#13421)

airbyte-commons-worker/src/main/kotlin/io/airbyte/workers/sync/OrchestratorConstants.kt

@@ -44,6 +44,7 @@ object OrchestratorConstants {
       // add variables defined in this file
       addAll(
         setOf(
+          EnvVar.FEATURE_FLAG_BASEURL.toString(),
           EnvVar.FEATURE_FLAG_CLIENT.toString(),
           EnvVar.FEATURE_FLAG_PATH.toString(),
           LOG_LEVEL,

airbyte-commons/src/main/kotlin/io/airbyte/commons/envvar/EnvVar.kt

@@ -35,6 +35,7 @@ enum class EnvVar {
   DOCKER_HOST,
   DOCKER_NETWORK,
 
+  FEATURE_FLAG_BASEURL,
   FEATURE_FLAG_CLIENT,
   FEATURE_FLAG_PATH,
 

airbyte-container-orchestrator/src/main/resources/application.yml

@@ -86,6 +86,7 @@ airbyte:
     client: ${FEATURE_FLAG_CLIENT:config}
     path: ${FEATURE_FLAG_PATH:/flags}
     api-key: ${LAUNCHDARKLY_KEY:}
+    base-url: ${FEATURE_FLAG_BASEURL:}
   internal-api:
     auth-header:
       name: ${AIRBYTE_API_AUTH_HEADER_NAME:}

airbyte-featureflag-server/Dockerfile

@@ -0,0 +1,16 @@
+ARG JDK_IMAGE=airbyte/airbyte-base-java-image:3.2.3
+
+FROM scratch as builder
+WORKDIR /app
+ADD airbyte-app.tar /app
+
+FROM ${JDK_IMAGE}
+EXPOSE 8007 5005
+ENV APPLICATION airbyte-featureflag-server
+ENV VERSION ${VERSION}
+
+WORKDIR /app
+COPY --chown=airbyte:airbyte --from=builder /app /app
+USER airbyte:airbyte
+
+ENTRYPOINT ["/bin/bash", "-c", "airbyte-app/bin/${APPLICATION}"]

airbyte-featureflag-server/build.gradle.kts

@@ -0,0 +1,61 @@
+import java.util.Properties
+
+plugins {
+  id("io.airbyte.gradle.jvm.app")
+  id("io.airbyte.gradle.docker")
+}
+
+dependencies {
+  ksp(libs.bundles.micronaut.annotation.processor)
+  ksp(libs.v3.swagger.annotations)
+  ksp(platform(libs.micronaut.platform))
+
+  annotationProcessor(libs.bundles.micronaut.annotation.processor)
+  annotationProcessor(libs.micronaut.jaxrs.processor)
+  annotationProcessor(platform(libs.micronaut.platform))
+
+  compileOnly(libs.v3.swagger.annotations)
+  compileOnly(libs.micronaut.openapi.annotations)
+
+  implementation(platform(libs.micronaut.platform))
+  implementation(libs.bundles.micronaut)
+  implementation(libs.bundles.micronaut.kotlin)
+  implementation(libs.log4j.impl)
+  implementation(libs.micronaut.http)
+  implementation(libs.micronaut.jaxrs.server)
+  implementation(libs.micronaut.security)
+  implementation(libs.v3.swagger.annotations)
+  implementation(libs.jackson.databind)
+  implementation(libs.jackson.dataformat)
+  implementation(libs.jackson.kotlin)
+
+  implementation(project(":oss:airbyte-commons"))
+
+  testImplementation(libs.bundles.micronaut.test)
+  testImplementation(libs.mockk)
+}
+
+val env = Properties().apply {
+  load(rootProject.file(".env.dev").inputStream())
+}
+
+airbyte {
+  application {
+    mainClass = "io.airbyte.featureflag.server.ApplicationKt"
+    defaultJvmArgs = listOf("-XX:+ExitOnOutOfMemoryError", "-XX:MaxRAMPercentage=75.0")
+    @Suppress("UNCHECKED_CAST")
+    localEnvVars.putAll(env.toMap() as Map<String, String>)
+    localEnvVars.putAll(
+      mapOf(
+        "AIRBYTE_ROLE" to (System.getenv("AIRBYTE_ROLE") ?: "undefined"),
+        "AIRBYTE_VERSION" to env["VERSION"].toString(),
+        "MICRONAUT_ENVIRONMENTS" to "control-plane",
+        "SERVICE_NAME" to project.name,
+        "TRACKING_STRATEGY" to env["TRACKING_STRATEGY"].toString(),
+      )
+    )
+  }
+  docker {
+    imageName = "featureflag-server"
+  }
+}

airbyte-featureflag-server/src/main/kotlin/io/airbyte/featureflag/server/Application.kt

@@ -0,0 +1,7 @@
+package io.airbyte.featureflag.server
+
+import io.micronaut.runtime.Micronaut.run
+
+fun main(args: Array<String>) {
+  run(*args)
+}

airbyte-featureflag-server/src/main/kotlin/io/airbyte/featureflag/server/FeatureFlagApi.kt

@@ -0,0 +1,150 @@
+package io.airbyte.featureflag.server
+
+import io.airbyte.featureflag.server.model.Context
+import io.airbyte.featureflag.server.model.FeatureFlag
+import io.airbyte.featureflag.server.model.Rule
+import io.micronaut.http.HttpStatus
+import io.micronaut.http.annotation.Body
+import io.micronaut.http.annotation.Consumes
+import io.micronaut.http.annotation.Controller
+import io.micronaut.http.annotation.PathVariable
+import io.micronaut.http.annotation.QueryValue
+import io.micronaut.scheduling.TaskExecutors
+import io.micronaut.scheduling.annotation.ExecuteOn
+import io.swagger.v3.oas.annotations.media.Content
+import io.swagger.v3.oas.annotations.media.Schema
+import io.swagger.v3.oas.annotations.parameters.RequestBody
+import io.swagger.v3.oas.annotations.responses.ApiResponse
+import io.swagger.v3.oas.annotations.responses.ApiResponses
+import jakarta.ws.rs.DELETE
+import jakarta.ws.rs.GET
+import jakarta.ws.rs.POST
+import jakarta.ws.rs.PUT
+import jakarta.ws.rs.Path
+
+@Controller("/api/v1/feature-flags")
+@ExecuteOn(TaskExecutors.IO)
+class FeatureFlagApi(private val ffs: FeatureFlagService) {
+  @DELETE
+  @Path("/{key}")
+  @ApiResponses(
+    value = [
+      ApiResponse(
+        responseCode = "200",
+        description = "The flag evaluation",
+      ),
+    ],
+  )
+  fun delete(
+    @PathVariable key: String,
+  ) {
+    ffs.delete(key)
+  }
+
+  @GET
+  @Path("/{key}")
+  @ApiResponses(
+    value = [
+      ApiResponse(
+        responseCode = "200",
+        description = "The FeatureFlag",
+        content = [Content(schema = Schema(implementation = FeatureFlag::class))],
+      ),
+    ],
+  )
+  fun get(
+    @PathVariable key: String,
+  ): FeatureFlag {
+    return ffs.get(key) ?: throw KnownException(HttpStatus.NOT_FOUND, "$key not found")
+  }
+
+  @PUT
+  @Path("/")
+  @Consumes("application/json")
+  @ApiResponses(
+    value = [
+      ApiResponse(
+        responseCode = "200",
+        description = "The FeatureFlag",
+        content = [Content(schema = Schema(implementation = FeatureFlag::class))],
+      ),
+    ],
+  )
+  fun put(
+    @RequestBody(content = [Content(schema = Schema(implementation = FeatureFlag::class))]) @Body request: FeatureFlag,
+  ): FeatureFlag {
+    return ffs.put(request)
+  }
+
+  @GET
+  @Path("/{key}/evaluate")
+  @ApiResponses(
+    value = [
+      ApiResponse(
+        responseCode = "200",
+        description = "The flag evaluation",
+      ),
+    ],
+  )
+  fun evaluate(
+    @PathVariable key: String,
+    @QueryValue("kind") kind: List<String> = emptyList(),
+    @QueryValue("value") value: List<String> = emptyList(),
+  ): String {
+    val context = kind.zip(value).toMap()
+    return ffs.eval(key, context) ?: throw KnownException(HttpStatus.NOT_FOUND, "$key not found")
+  }
+
+  @DELETE
+  @Path("/{key}/rules")
+  @ApiResponses(
+    value = [
+      ApiResponse(
+        responseCode = "200",
+        description = "The flag evaluation",
+      ),
+    ],
+  )
+  fun deleteRule(
+    @PathVariable key: String,
+    @RequestBody(content = [Content(schema = Schema(implementation = Context::class))]) @Body context: Context,
+  ): FeatureFlag {
+    return ffs.removeRule(key, context)
+  }
+
+  @POST
+  @Path("/{key}/rules")
+  @ApiResponses(
+    value = [
+      ApiResponse(
+        responseCode = "200",
+        description = "The flag evaluation",
+      ),
+    ],
+  )
+  fun postRule(
+    @PathVariable key: String,
+    @RequestBody(content = [Content(schema = Schema(implementation = Rule::class))]) @Body rule: Rule,
+  ): FeatureFlag {
+    return ffs.addRule(key, rule)
+  }
+
+  @PUT
+  @Path("/{key}/rules")
+  @ApiResponses(
+    value = [
+      ApiResponse(
+        responseCode = "200",
+        description = "The flag evaluation",
+      ),
+    ],
+  )
+  fun putRule(
+    @PathVariable key: String,
+    @RequestBody(content = [Content(schema = Schema(implementation = Rule::class))]) @Body rule: Rule,
+  ): FeatureFlag {
+    return ffs.updateRule(key, rule)
+  }
+}
+
+class KnownException(val httpCode: HttpStatus, message: String) : Throwable(message)

airbyte-featureflag-server/src/main/kotlin/io/airbyte/featureflag/server/FeatureFlagService.kt

@@ -0,0 +1,136 @@
+package io.airbyte.featureflag.server
+
+import com.fasterxml.jackson.databind.ObjectMapper
+import com.fasterxml.jackson.dataformat.yaml.YAMLFactory
+import com.fasterxml.jackson.module.kotlin.registerKotlinModule
+import io.airbyte.featureflag.server.model.Context
+import io.airbyte.featureflag.server.model.FeatureFlag
+import io.airbyte.featureflag.server.model.Rule
+import io.micronaut.context.annotation.Property
+import jakarta.inject.Singleton
+import java.nio.file.Path
+import kotlin.io.path.exists
+import kotlin.io.path.isRegularFile
+
+// This is open for testing, creating an interface might be the way to go
+@Singleton
+open class FeatureFlagService(
+  @Property(name = "airbyte.feature-flag.path") configPath: Path?,
+) {
+  private val flags = mutableMapOf<String, MutableFeatureFlag>()
+
+  init {
+    configPath?.also { path ->
+      if (path.exists() && path.isRegularFile()) {
+        val yamlMapper = ObjectMapper(YAMLFactory()).registerKotlinModule()
+        val config = yamlMapper.readValue(path.toFile(), ConfigFileFlags::class.java)
+        config.flags.forEach {
+          put(it.toFeatureFlag())
+        }
+      }
+    }
+  }
+
+  open fun delete(key: String) {
+    flags.remove(key)
+  }
+
+  open fun eval(
+    key: String,
+    context: Map<String, String>,
+  ): String? {
+    val flag = flags[key] ?: return null
+    for (rule in flag.rules) {
+      if (rule.context.matches(context)) {
+        return rule.value
+      }
+    }
+    return flag.default
+  }
+
+  open fun get(key: String): FeatureFlag? {
+    return flags[key]?.toFeatureFlag()
+  }
+
+  open fun addRule(
+    key: String,
+    rule: Rule,
+  ): FeatureFlag {
+    val flag = flags[key] ?: throw Exception("$key not found")
+
+    if (flag.rules.any { it.context == rule.context }) {
+      throw Exception("$key already has a rule for context ${rule.context}")
+    }
+    flag.rules.add(rule.toMutableRule())
+    return flag.toFeatureFlag()
+  }
+
+  open fun updateRule(
+    key: String,
+    rule: Rule,
+  ): FeatureFlag {
+    val flag = flags[key] ?: throw Exception("$key not found")
+    flag.rules
+      .find { it.context == rule.context }
+      ?.apply { value = rule.value }
+      ?: throw Exception("$key does not have a rule for context ${rule.context}")
+    return flag.toFeatureFlag()
+  }
+
+  open fun removeRule(
+    key: String,
+    context: Context,
+  ): FeatureFlag {
+    val flag = flags[key] ?: throw Exception("$key not found")
+    flag.rules.removeIf { it.context == context }
+    return flag.toFeatureFlag()
+  }
+
+  open fun put(flag: FeatureFlag): FeatureFlag {
+    flags[flag.key] = flag.toMutableFeatureFlag()
+    return get(flag.key) ?: throw Exception("Failed to put flag $flag")
+  }
+
+  open fun put(
+    key: String,
+    default: String,
+    rules: List<Rule> = emptyList(),
+  ): FeatureFlag {
+    val flag = FeatureFlag(key = key, default = default, rules = rules)
+    return put(flag)
+  }
+
+  private fun Context.matches(env: Map<String, String>): Boolean = env[kind] == value
+
+  private fun MutableFeatureFlag.toFeatureFlag(): FeatureFlag = FeatureFlag(key = key, default = default, rules = rules.map { it.toRule() }.toList())
+
+  private fun FeatureFlag.toMutableFeatureFlag(): MutableFeatureFlag =
+    MutableFeatureFlag(key = key, default = default, rules = rules.map { it.toMutableRule() }.toMutableList())
+
+  private fun MutableRule.toRule(): Rule = Rule(context = context, value = value)
+
+  private fun Rule.toMutableRule(): MutableRule = MutableRule(context = context, value = value)
+
+  private data class MutableFeatureFlag(val key: String, val default: String, val rules: MutableList<MutableRule>)
+
+  private data class MutableRule(val context: Context, var value: String)
+
+  private data class ConfigFileFlags(val flags: List<ConfigFileFlag>)
+
+  private data class ConfigFileFlag(
+    val name: String,
+    val serve: String,
+    val context: List<ConfigFileFlagContext>? = null,
+  ) {
+    fun toFeatureFlag(): FeatureFlag {
+      val rules = context?.flatMap { context -> context.include.map { Rule(Context(kind = context.type, value = it), value = context.serve) } }
+      return FeatureFlag(key = name, default = serve, rules = rules ?: emptyList())
+    }
+  }
+
+  private data class ConfigFileFlagContext(
+    val type: String,
+    val serve: String,
+    val include: List<String> = listOf(),
+  )
+}