feat(pace-69): setup plugin system and add openai gpt4 based data policy generator (#91)

Co-authored-by: Ivan Plantevin <[email protected]>
Co-authored-by: Bob van den Hoogen <[email protected]>

Author: 3 people

.github/workflows/publish-release-artifacts.yaml

@@ -42,6 +42,17 @@ jobs:
           buf beta registry tag create "buf.build/getstrm/pace:${buf_commit_hash}" ${{ env.TAG }}
           buf_dependency_version=$(buf alpha sdk maven-version --module=buf.build/getstrm/pace:${{ env.TAG }} --plugin=buf.build/protocolbuffers/java:v24.4 | sed "s/24.4.0.1.//")
           echo "BUF_DEPENDENCY_VERSION=${buf_dependency_version}" >> "$GITHUB_ENV"
+      - name: Generate JSON Schema
+        env:
+          BUF_TOKEN: ${{ secrets.BUF_TOKEN }}
+        run: |
+          cd protos
+          buf generate
+      - name: Copy JSON Schema
+        run: |
+          find app/src/main/resources/jsonschema -type d -maxdepth 1 -mindepth 1 | xargs -I{} rm -rf {}
+          mkdir -p app/src/main/resources/jsonschema
+          cp -r protos/json-schema/* app/src/main/resources/jsonschema
 
       # JVM artifact
       - uses: actions/setup-java@v3

.gitignore

@@ -17,3 +17,6 @@ protos/openapi.yaml
 integration-test/pace.log
 integration-test/awk
 integration-test/cut
+app/src/test/resources/openai.properties
+
+rest/envoy-local.yaml

Makefile

@@ -6,10 +6,10 @@ SHELL := /bin/bash
 git_branch := $(shell git rev-parse --abbrev-ref HEAD)
 descriptor_file := "rest/descriptor.binpb"
 
-buf-publish-current-branch:
-	[[ "$$OSTYPE" == "darwin"* ]] && SED=gsed || SED=sed && \
-	commit_hash=$$(cd protos > /dev/null && buf push --branch "${git_branch}") && \
-	[ ! -z "$$commit_hash" ] && commit_hash_short=$$(echo "$$commit_hash" | cut -c1-12) && $$SED -i "s|generatedBufDependencyVersion=.*|generatedBufDependencyVersion=00000000000000.$$commit_hash_short|g" gradle.properties || echo "No changes to protos, gradle.properties not updated"
+buf-publish-current-branch: copy-json-schema-to-resources
+	@ [[ "$$OSTYPE" == "darwin"* ]] && SED=gsed || SED=sed
+	@ commit_hash=$$(cd protos > /dev/null && buf push --branch "${git_branch}")
+	@ [ ! -z "$$commit_hash" ] && commit_hash_short=$$(echo "$$commit_hash" | cut -c1-12) && $$SED -i "s|generatedBufDependencyVersion=.*|generatedBufDependencyVersion=00000000000000.$$commit_hash_short|g" gradle.properties || echo "No changes to protos, gradle.properties not updated"
 
 run-docker-local:
 	./gradlew buildDocker && docker run -p 8080:8080 -p 9090:9090 -p 50051:50051 -e SPRING_PROFILES_ACTIVE=dockerdev pace:latest
@@ -18,7 +18,11 @@ buf-create-descriptor-binpb: # PHONY on purpose, as we want to regenerate every
 	rm -f ${descriptor_file} && \
 	buf build protos --config ./protos/buf.yaml -o ${descriptor_file}
 
-run-rest-proxy: buf-create-descriptor-binpb
+create-envoy-spec:
+	@ export GRPC_SERVICES=$$(buf build -o -#format=json | jq -rc '.file | map(select(.name | startswith("getstrm"))) | map(select(.service > 0) | (.package + "." + .service[].name))')
+	@ envsubst < rest/envoy-local-template.yaml > rest/envoy-local.yaml
+
+run-rest-proxy: buf-create-descriptor-binpb create-envoy-spec
 	docker run -p 9090:9090 -p 9000:9000 -v $$(pwd)/rest/envoy-local.yaml:/etc/envoy/envoy.yaml -v $$(pwd)/${descriptor_file}:/tmp/envoy/descriptor.binpb envoyproxy/envoy:v1.28-latest
 
 /tmp/envoy.yaml: rest/envoy-local.yaml
@@ -28,4 +32,10 @@ run-rest-proxy-localhost: buf-create-descriptor-binpb /tmp/envoy.yaml
 	docker run --net=host -p 9090:9090 -p 9000:9000 -v /tmp/envoy.yaml:/etc/envoy/envoy.yaml -v $$(pwd)/${descriptor_file}:/tmp/envoy/descriptor.binpb envoyproxy/envoy:v1.28-latest
 
 json-schema:
-	(cd protos; buf generate)
+	@ rm -rf protos/json-schema
+	@ (cd protos; buf generate)
+
+copy-json-schema-to-resources: json-schema
+	@ find app/src/main/resources/jsonschema -type d -maxdepth 1 -mindepth 1 | xargs -I{} rm -rf {}
+	@ mkdir -p app/src/main/resources/jsonschema
+	@ cp -r protos/json-schema/* app/src/main/resources/jsonschema

app/build.gradle.kts

@@ -6,6 +6,7 @@ import nu.studer.gradle.jooq.JooqGenerate
 import org.flywaydb.gradle.task.FlywayMigrateTask
 import org.gradle.api.tasks.testing.logging.TestExceptionFormat
 import org.springframework.boot.gradle.tasks.bundling.BootJar
+import java.io.ByteArrayOutputStream
 import java.net.InetAddress
 import java.net.Socket
 import java.time.Instant
@@ -21,6 +22,7 @@ val postgresPort: Int by rootProject.extra
 val jooqSchema = rootProject.extra["schema"] as String
 val jooqMigrationDir = "$projectDir/src/main/resources/db/migration/postgresql"
 val jooqVersion = rootProject.ext["jooqVersion"] as String
+val kotestVersion = rootProject.ext["kotestVersion"] as String
 val openDataDiscoveryOpenApiDir = layout.buildDirectory.dir("generated/source/odd").get()
 project.version = if (gradle.startParameter.taskNames.any { it.lowercase() == "builddocker" }) {
     "${project.version}-SNAPSHOT"
@@ -44,7 +46,7 @@ dependencies {
     // Dependencies managed by Spring
     implementation("org.springframework.boot:spring-boot-starter-jooq")
     // TODO remove once we upgrade Spring: override SnakeYAML dependency, as the one managed by Spring is too old and is vulnerable
-    implementation("org.yaml:snakeyaml:2.0")
+    implementation("org.yaml:snakeyaml:2.2")
     implementation("org.flywaydb:flyway-core")
     implementation("org.jetbrains.kotlin:kotlin-reflect")
     implementation("org.postgresql:postgresql")
@@ -60,28 +62,35 @@ dependencies {
 
     // Self-managed dependencies
     implementation("net.devh:grpc-server-spring-boot-starter:2.15.0.RELEASE")
-    implementation("com.databricks:databricks-sdk-java:0.10.0")
+    implementation("com.databricks:databricks-sdk-java:0.13.0")
     implementation("com.github.drapostolos:type-parser:0.8.1")
-    implementation("com.microsoft.sqlserver:mssql-jdbc:12.4.2.jre8")
+    implementation("com.microsoft.sqlserver:mssql-jdbc:12.4.2.jre11")
 
-    implementation("com.nimbusds:nimbus-jose-jwt:9.37")
-    implementation("org.bouncycastle:bcpkix-jdk18on:1.76")
+    implementation("com.nimbusds:nimbus-jose-jwt:9.37.1")
+    implementation("org.bouncycastle:bcpkix-jdk18on:1.77")
 
-    implementation(enforcedPlatform("com.google.cloud:libraries-bom:26.24.0"))
+    implementation(enforcedPlatform("com.google.cloud:libraries-bom:26.27.0"))
     implementation("com.google.cloud:google-cloud-bigquery")
     implementation("com.google.cloud:google-cloud-datacatalog")
 
     implementation("build.buf.gen:getstrm_pace_grpc_java:1.59.0.2.$generatedBufDependencyVersion")
     implementation("build.buf.gen:getstrm_pace_grpc_kotlin:1.4.1.1.$generatedBufDependencyVersion")
-    implementation("build.buf.gen:getstrm_pace_protocolbuffers_java:24.4.0.1.$generatedBufDependencyVersion")
-    implementation("build.buf:protovalidate:0.1.6")
+    implementation("build.buf.gen:getstrm_pace_protocolbuffers_java:25.1.0.1.$generatedBufDependencyVersion")
+    implementation("build.buf:protovalidate:0.1.8")
 
     implementation("com.apollographql.apollo3:apollo-runtime:3.8.2")
 
+    implementation("com.aallam.openai:openai-client:3.6.1")
+    implementation(platform("io.ktor:ktor-bom:2.3.6"))
+    runtimeOnly("io.ktor:ktor-client-okhttp")
+    implementation("io.ktor:ktor-client-logging")
+
     // Test dependencies
-    testImplementation("org.springframework.boot:spring-boot-starter-test")
-    testImplementation("io.kotest:kotest-assertions-core-jvm:5.7.2")
-    testImplementation("io.kotest:kotest-runner-junit5:5.7.2")
+    testImplementation("org.springframework.boot:spring-boot-starter-test") {
+        exclude(group = "com.vaadin.external.google", module = "android-json")
+    }
+    testImplementation("io.kotest:kotest-assertions-core-jvm:$kotestVersion")
+    testImplementation("io.kotest:kotest-runner-junit5:$kotestVersion")
     testImplementation("io.mockk:mockk:1.13.8")
     testImplementation("io.zonky.test:embedded-postgres:2.0.4")
 }
@@ -308,11 +317,27 @@ val createProtoDescriptor =
 val copyDocker =
     tasks.register<Copy>("copyDocker") {
         group = "docker"
+        val grpcServices: String = ByteArrayOutputStream().use { outputStream ->
+            project.exec {
+                workingDir("$rootDir/protos")
+
+                commandLine(
+                    "bash",
+                    "-c",
+                    "buf build -o -#format=json | jq -rc '.file | map(select(.name | startswith(\"getstrm\"))) | map(select(.service > 0) | (.package + \".\" + .service[].name))'"
+                )
+                standardOutput = outputStream
+            }
+
+            outputStream.toString()
+        }
+
         from("src/main/docker")
         include("*")
         into("build/docker")
         expand(
             "version" to project.version,
+            "grpcServices" to grpcServices,
         )
     }
 

app/src/main/docker/envoy.yaml

@@ -29,15 +29,7 @@ static_resources:
                     typed_config:
                       "@type": type.googleapis.com/envoy.extensions.filters.http.grpc_json_transcoder.v3.GrpcJsonTranscoder
                       proto_descriptor: "/app/descriptor.binpb"
-                      # TODO Leaving this hardcoded for now, as we only have a few service at the moment
-                      # In order to dynamically determine the list of services, we can use the following command
-                      # buf build -o -#format=json | jq -r '.file | map(select(.name | startswith("getstrm"))) | map(select(.service > 0) | (.package + "." + .service[].name))[] | ("- " + .)'
-                      # With the `copyDocker` command in the gradle build file, this could be expanded as a variable in this file
-                      services:
-                      - "getstrm.pace.api.data_catalogs.v1alpha.DataCatalogsService"
-                      - "getstrm.pace.api.data_policies.v1alpha.DataPoliciesService"
-                      - "getstrm.pace.api.processing_platforms.v1alpha.ProcessingPlatformsService"
-                      - "getstrm.pace.api.global_transforms.v1alpha.GlobalTransformsService"
+                      services: ${grpcServices}
                       auto_mapping: false
                       convert_grpc_status: true
                       request_validation_options:

app/src/main/kotlin/com/getstrm/pace/api/PluginsApi.kt

@@ -0,0 +1,21 @@
+package com.getstrm.pace.api
+
+import build.buf.gen.getstrm.pace.api.plugins.v1alpha.*
+import com.getstrm.pace.service.PluginsService
+import net.devh.boot.grpc.server.service.GrpcService
+
+@GrpcService
+class PluginsApi(private val pluginsService: PluginsService) : PluginsServiceGrpcKt.PluginsServiceCoroutineImplBase() {
+    override suspend fun listPlugins(request: ListPluginsRequest): ListPluginsResponse =
+        ListPluginsResponse.newBuilder()
+            .addAllPlugins(pluginsService.listPlugins())
+            .build()
+
+    override suspend fun getPayloadJSONSchema(request: GetPayloadJSONSchemaRequest): GetPayloadJSONSchemaResponse =
+        GetPayloadJSONSchemaResponse.newBuilder()
+            .setSchema(pluginsService.getPluginPayloadJsonSchema(request.pluginId))
+            .build()
+
+    override suspend fun invokePlugin(request: InvokePluginRequest): InvokePluginResponse =
+        pluginsService.invokePlugin(request)
+}

app/src/main/kotlin/com/getstrm/pace/config/AppConfiguration.kt

@@ -2,9 +2,6 @@ package com.getstrm.pace.config
 
 import build.buf.gen.getstrm.pace.api.entities.v1alpha.DataCatalog
 import org.springframework.boot.context.properties.ConfigurationProperties
-import org.springframework.boot.context.properties.ConfigurationPropertiesScan
-import org.springframework.boot.context.properties.EnableConfigurationProperties
-import org.springframework.stereotype.Component
 
 @ConfigurationProperties(prefix = "app")
 data class AppConfiguration(
@@ -21,4 +18,3 @@ data class CatalogConfiguration(
     val password: String?,
     val fetchSize: Int? = 1,
 )
-

app/src/main/kotlin/com/getstrm/pace/config/PaceConfiguration.kt

@@ -3,14 +3,12 @@ package com.getstrm.pace.config
 import com.getstrm.pace.grpc.ExceptionHandlerInterceptor
 import com.getstrm.pace.grpc.ValidationInterceptor
 import net.devh.boot.grpc.server.interceptor.GrpcGlobalServerInterceptor
+import org.springframework.boot.context.properties.ConfigurationPropertiesScan
 import org.springframework.boot.context.properties.EnableConfigurationProperties
 import org.springframework.context.annotation.Configuration
 
 @Configuration
-@EnableConfigurationProperties(ProcessingPlatformConfiguration::class,
-    AppConfiguration::class,
-    GlobalTransformsConfiguration::class,
-    )
+@ConfigurationPropertiesScan("com.getstrm.pace.config")
 class PaceConfiguration {
     @GrpcGlobalServerInterceptor
     fun exceptionInterceptor() = ExceptionHandlerInterceptor(false)

app/src/main/kotlin/com/getstrm/pace/exceptions/ProtoValidator.kt

@@ -0,0 +1,49 @@
+package com.getstrm.pace.exceptions
+
+import build.buf.protovalidate.ValidationResult
+import build.buf.protovalidate.Validator
+import build.buf.protovalidate.exceptions.ValidationException
+import com.google.protobuf.Message
+import com.google.rpc.BadRequest
+import com.google.rpc.DebugInfo
+import org.slf4j.LoggerFactory
+
+object ProtoValidator {
+    private val log by lazy { LoggerFactory.getLogger(javaClass) }
+    private val validator = Validator()
+
+    fun validate(message: Message): PaceStatusException? {
+        return try {
+            val result: ValidationResult = validator.validate(message)
+
+            if (result.violations.isEmpty()) {
+                null
+            } else {
+                val violations = result.violations.map {
+                    BadRequest.FieldViolation.newBuilder()
+                        .setField(if (it.forKey) "${it.fieldPath} (map key)" else it.fieldPath)
+                        .setDescription("${it.message} (constraint id = ${it.constraintId})")
+                        .build()
+                }
+
+                BadRequestException(
+                    BadRequestException.Code.INVALID_ARGUMENT,
+                    BadRequest.newBuilder()
+                        .addAllFieldViolations(violations)
+                        .build(),
+                    errorMessage = "Validation of message ${message.descriptorForType.name} failed, ${violations.size} violations found",
+                )
+            }
+        } catch (e: ValidationException) {
+            val protoMessageName = message.descriptorForType.name ?: "'Cannot determine Proto Message name'"
+            log.error("An exception occurred while validating message $protoMessageName", e)
+            InternalException(
+                InternalException.Code.INTERNAL,
+                DebugInfo.newBuilder()
+                    .setDetail("Validation of message $protoMessageName failed, ${e.message}, ${PaceStatusException.BUG_REPORT}")
+                    .addAllStackEntries(e.stackTrace.map { it.toString() })
+                    .build()
+            )
+        }
+    }
+}