Flex template deployment after merge to main

.github/workflows/ci.yml

@@ -50,7 +50,7 @@ jobs:
 
   test:
     runs-on: ubuntu-latest
-    needs: [scalafmt, scapegoat, scalafix]
+    needs: [scalafmt, scapegoat, scalafix, tf-validate, tf-fmt]
     steps:
       - uses: actions/checkout@v4
       - uses: ./.github/actions/setup

.github/workflows/deploy.yml

@@ -0,0 +1,28 @@
+name: Deploy
+
+on:
+  workflow_dispatch: ~
+  workflow_run:
+    workflows: ["CI"]
+    types: [completed]
+    branches:
+      - "main"
+
+jobs:
+  deploy:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+      - uses: ./.github/actions/setup
+      - run: sbt tollApplication/assembly
+      - uses: google-github-actions/auth@v1
+        with:
+          credentials_json: "${{ secrets.GOOGLE_CREDENTIALS }}"
+      - uses: dflook/terraform-apply@v1
+        with:
+          path: toll-infrastructure
+      - uses: "google-github-actions/setup-gcloud@v1"
+      - run: ./deploy-batch.sh
+        working-directory: toll-application
+      - run: ./deploy-streaming.sh
+        working-directory: toll-application

build.sbt

@@ -18,6 +18,8 @@ lazy val test = (project in file("stream-processing-test"))
       scio,
       scioGcp,
       scioTest,
+      beamDirectRunner,
+      beamDataflowRunner,
       scalaLogging,
       slf4j,
       slf4jJcl,
@@ -35,6 +37,8 @@ lazy val shared = (project in file("stream-processing-shared"))
     libraryDependencies ++= Seq(
       scio,
       scioGcp,
+      beamDirectRunner,
+      beamDataflowRunner,
       scalaLogging,
       slf4j,
       slf4jJcl,
@@ -49,6 +53,8 @@ lazy val infrastructure = (project in file("stream-processing-infrastructure"))
     libraryDependencies ++= Seq(
       scio,
       scioGcp,
+      beamDirectRunner,
+      beamDataflowRunner,
       scalaLogging,
       slf4j,
       slf4jJcl,
@@ -82,7 +88,11 @@ lazy val tollDomain = (project in file("toll-domain"))
   )
 
 lazy val tollApplication = (project in file("toll-application"))
-  .settings(commonSettings)
+  .settings(
+    commonSettings,
+    assemblySettings,
+    assembly / assemblyJarName := "toll-application.jar"
+  )
   .dependsOn(
     shared,
     infrastructure,

project/Dependencies.scala

@@ -5,6 +5,9 @@ object Dependencies {
   val scioGcp = "com.spotify" %% "scio-google-cloud-platform" % "0.13.3"
   val scioTest = "com.spotify" %% "scio-test" % "0.13.3"
 
+  val beamDirectRunner = "org.apache.beam" % "beam-runners-direct-java" % "2.50.0"
+  val beamDataflowRunner = "org.apache.beam" % "beam-runners-google-cloud-dataflow-java" % "2.6.0"
+
   val scalaLogging = "com.typesafe.scala-logging" %% "scala-logging" % "3.9.5"
 
   val slf4j = "org.slf4j" % "slf4j-api" % "2.0.9"

project/Settings.scala

@@ -6,6 +6,7 @@ import sbt.Keys._
 import com.github.sbt.jacoco.JacocoKeys.jacocoReportSettings
 import com.github.sbt.jacoco.JacocoKeys.JacocoReportFormats
 import com.github.sbt.jacoco.JacocoPlugin.autoImport.JacocoReportSettings
+import sbtassembly.AssemblyPlugin.autoImport._
 import scalafix.sbt.ScalafixPlugin.autoImport._
 
 object Settings {
@@ -47,4 +48,19 @@ object Settings {
     ThisBuild / semanticdbVersion := scalafixSemanticdb.revision,
     ThisBuild / scalafixScalaBinaryVersion := CrossVersion.binaryScalaVersion(scalaVersion.value)
   )
+
+  val assemblySettings = Seq(
+    assembly / assemblyMergeStrategy := {
+      case s if s.endsWith(".class")      => MergeStrategy.last
+      case s if s.endsWith(".proto")      => MergeStrategy.last
+      case s if s.endsWith(".properties") => MergeStrategy.filterDistinctLines
+      case PathList("META-INF", "gradle", "incremental.annotation.processors") =>
+        MergeStrategy.discard
+      case PathList("git.properties") =>
+        MergeStrategy.discard
+      case x =>
+        val oldStrategy = (ThisBuild / assemblyMergeStrategy).value
+        oldStrategy(x)
+    }
+  )
 }

project/plugins.sbt

@@ -7,3 +7,5 @@ addSbtPlugin("org.scalameta" % "sbt-scalafmt" % "2.5.2")
 addSbtPlugin("com.sksamuel.scapegoat" %% "sbt-scapegoat" % "1.2.2")
 // automated code changes
 addSbtPlugin("ch.epfl.scala" % "sbt-scalafix" % "0.11.1")
+// fat-jar for flex templates
+addSbtPlugin("com.eed3si9n" % "sbt-assembly" % "2.1.3")

toll-application/deploy-batch.sh

@@ -0,0 +1,11 @@
+#!/bin/bash
+
+PROJECT=playground-272019
+REGION=europe-west1
+
+gcloud dataflow flex-template build gs://$PROJECT-toll-application/templates/toll-application-batch.json \
+    --image-gcr-path "$REGION-docker.pkg.dev/$PROJECT/toll-application/toll-application-batch:latest" \
+    --sdk-language "JAVA" \
+    --flex-template-base-image JAVA17 \
+    --jar "target/scala-2.13/toll-application.jar" \
+    --env FLEX_TEMPLATE_JAVA_MAIN_CLASS="org.mkuthan.streamprocessing.toll.application.batch.TollBatchJob"

toll-application/deploy-streaming.sh

@@ -0,0 +1,11 @@
+#!/bin/bash
+
+PROJECT=playground-272019
+REGION=europe-west1
+
+gcloud dataflow flex-template build gs://$PROJECT-toll-application/templates/toll-application-streaming.json \
+    --image-gcr-path "$REGION-docker.pkg.dev/$PROJECT/toll-application/toll-application-streaming:latest" \
+    --sdk-language "JAVA" \
+    --flex-template-base-image JAVA17 \
+    --jar "target/scala-2.13/toll-application.jar" \
+    --env FLEX_TEMPLATE_JAVA_MAIN_CLASS="org.mkuthan.streamprocessing.toll.application.streaming.TollStreamingJob"

toll-application/run-streaming.sh

@@ -0,0 +1,22 @@
+#!/bin/bash
+
+PROJECT=playground-272019
+REGION=europe-west1
+
+gcloud dataflow flex-template run "toll-application-`date +%Y%m%d-%H%M%S`" \
+    --template-file-gcs-location "gs://$PROJECT-toll-application/templates/toll-application-streaming.json" \
+    --region "$REGION" \
+    --enable-streaming-engine \
+    --parameters entrySubscription="projects/$PROJECT/subscriptions/toll-booth-entry" \
+    --parameters entryDlq="gs://$PROJECT-toll-application/dlq/entry" \
+    --parameters exitSubscription="projects/$PROJECT/subscriptions/toll-booth-exit" \
+    --parameters exitDlq="gs://$PROJECT-toll-application/dlq/exit" \
+    --parameters vehicleRegistrationSubscription="projects/$PROJECT/subscriptions/vehicle-registration" \
+    --parameters vehicleRegistrationDlq="gs://$PROJECT-toll-application/dlq/vehicle-registration" \
+    --parameters vehicleRegistrationTable="$PROJECT.toll_application.vehicle-registration" \
+    --parameters entryStatsTable="$PROJECT.toll_application.toll-booth-entry-stats" \
+    --parameters totalVehicleTimesTable="$PROJECT.toll_application.total-vehicle-times" \
+    --parameters totalVehicleTimesDiagnosticTable="$PROJECT.toll_application.total-vehicle-times-diagnostic" \
+    --parameters vehiclesWithExpiredRegistrationTopic="projects/$PROJECT/topics/vehicle-registration" \
+    --parameters vehiclesWithExpiredRegistrationDiagnosticTable="$PROJECT.toll_application.vehicles-with-expired-registration-diagnostic" \
+    --parameters ioDiagnosticTable="$PROJECT.toll_application.io-diagnostic"

toll-infrastructure/main.tf

@@ -14,10 +14,13 @@ terraform {
 
 provider "google" {
   project = "playground-272019"
-  region  = "eu-west1"
-  zone    = "eu-west1-a"
+  region  = "europe-west1"
 }
 
+resource "google_artifact_registry_repository" "toll-application-registry-repository" {
+  repository_id = "toll-application"
+  format        = "DOCKER"
+}
 resource "google_storage_bucket" "toll-application-bucket" {
   name     = "playground-272019-toll-application"
   location = "EU"
@@ -58,27 +61,31 @@ resource "google_bigquery_table" "vehicle-registration-table" {
   table_id   = "vehicle-registration"
   dataset_id = google_bigquery_dataset.toll-application-dataset.dataset_id
 
+  # TODO: define schema
   schema = file("${path.module}/schemas/vehicle-registration.json")
 }
 
 resource "google_bigquery_table" "toll-booth-entry-stats-table" {
   table_id   = "toll-booth-entry-stats"
   dataset_id = google_bigquery_dataset.toll-application-dataset.dataset_id
 
+  # TODO: define schema
   schema = file("${path.module}/schemas/toll-booth-entry-stats.json")
 }
 
 resource "google_bigquery_table" "total-vehicle-times-table" {
   table_id   = "total-vehicle-times"
   dataset_id = google_bigquery_dataset.toll-application-dataset.dataset_id
 
+  # TODO: define schema
   schema = file("${path.module}/schemas/total-vehicle-times.json")
 }
 
 resource "google_bigquery_table" "total-vehicle-times-diagnostic-table" {
   table_id   = "total-vehicle-times-diagnostic"
   dataset_id = google_bigquery_dataset.toll-application-dataset.dataset_id
 
+  # TODO: define schema
   schema = file("${path.module}/schemas/toll-booth-diagnostic.json")
 }
 
@@ -90,12 +97,14 @@ resource "google_bigquery_table" "vehicles-with-expired-registration-diagnostic-
   table_id   = "vehicles-with-expired-registration-diagnostic"
   dataset_id = google_bigquery_dataset.toll-application-dataset.dataset_id
 
+  # TODO: define schema
   schema = file("${path.module}/schemas/toll-booth-diagnostic.json")
 }
 
 resource "google_bigquery_table" "io-diagnostic-table" {
   table_id   = "io-diagnostic"
   dataset_id = google_bigquery_dataset.toll-application-dataset.dataset_id
 
+  # TODO: define schema
   schema = file("${path.module}/schemas/io-diagnostic.json")
 }