Set up Flink SQL Gateway (#85)

Author: jogrogan

.github/workflows/integration-tests.yml

@@ -32,6 +32,7 @@ jobs:
       run: |
         kind load docker-image hoptimator
         kind load docker-image hoptimator-flink-runner
+        kind load docker-image hoptimator-flink-operator
     - name: Deploy Dev Environment
       run: make deploy-dev-environment
     - name: Deploy Hoptimator
@@ -64,4 +65,4 @@ jobs:
     - name: Capture Flink Job Logs
       if: always()
       run: kubectl logs $(kubectl get pods -l app.kubernetes.io/name=flink-kubernetes-operator -o name)
- 
+

Makefile

@@ -8,7 +8,8 @@ test:
 build:
 	./gradlew build
 	docker build . -t hoptimator
-	docker build hoptimator-flink-runner -t hoptimator-flink-runner
+	docker build hoptimator-flink-runner -f hoptimator-flink-runner/Dockerfile-flink-runner -t hoptimator-flink-runner
+	docker build hoptimator-flink-runner -f hoptimator-flink-runner/Dockerfile-flink-operator -t hoptimator-flink-operator
 
 bounce: build undeploy deploy deploy-samples deploy-config deploy-demo
 
@@ -47,26 +48,32 @@ deploy-samples: deploy
 undeploy-samples: undeploy
 	kubectl delete -f ./deploy/samples || echo "skipping"
 
-deploy-flink:
+deploy-flink: deploy
+	kubectl create -f https://github.com/jetstack/cert-manager/releases/download/v1.8.2/cert-manager.yaml || echo "skipping"
 	helm repo add flink-operator-repo https://downloads.apache.org/flink/flink-kubernetes-operator-1.9.0/
-	helm upgrade --install --atomic --set webhook.create=false flink-kubernetes-operator flink-operator-repo/flink-kubernetes-operator
+	helm upgrade --install --atomic --set webhook.create=false,image.pullPolicy=Never,image.repository=docker.io/library/hoptimator-flink-operator,image.tag=latest flink-kubernetes-operator flink-operator-repo/flink-kubernetes-operator
+	kubectl apply -f deploy/samples/flinkDeployment.yaml
+	kubectl apply -f deploy/samples/flinkSessionJob.yaml
+	docker compose -f ./deploy/docker/flink/docker-compose-sql-gateway.yaml up -d --wait
 
 undeploy-flink:
-	kubectl delete flinkdeployments.flink.apache.org --all || echo "skipping"
+	docker compose -f ./deploy/docker/flink/docker-compose-sql-gateway.yaml down
 	kubectl delete flinksessionjobs.flink.apache.org --all || echo "skipping"
-	kubectl delete crd flinkdeployments.flink.apache.org || echo "skipping"
+	kubectl delete flinkdeployments.flink.apache.org --all || echo "skipping"
 	kubectl delete crd flinksessionjobs.flink.apache.org || echo "skipping"
+	kubectl delete crd flinkdeployments.flink.apache.org || echo "skipping"
 	helm uninstall flink-kubernetes-operator || echo "skipping"
+	helm repo remove flink-operator-repo || echo "skipping"
 
 deploy-kafka: deploy deploy-flink
-	kubectl create -f https://github.com/jetstack/cert-manager/releases/download/v1.8.2/cert-manager.yaml || echo "skipping"
 	kubectl create namespace kafka || echo "skipping"
 	kubectl apply -f "https://strimzi.io/install/latest?namespace=kafka" -n kafka
 	kubectl wait --for=condition=Established=True crds/kafkas.kafka.strimzi.io
 	kubectl apply -f ./hoptimator-k8s/src/main/resources/
 	kubectl apply -f ./deploy/dev
 	kubectl apply -f ./deploy/samples/demodb.yaml
 	kubectl apply -f ./deploy/samples/kafkadb.yaml
+	kubectl delete -f https://github.com/jetstack/cert-manager/releases/download/v1.8.2/cert-manager.yaml || echo "skipping"
 
 undeploy-kafka:
 	kubectl delete kafkatopic.kafka.strimzi.io -n kafka --all || echo "skipping"
@@ -78,18 +85,17 @@ undeploy-kafka:
 	kubectl delete -f ./deploy/dev || echo "skipping"
 	kubectl delete -f ./hoptimator-k8s/src/main/resources/ || echo "skipping"
 	kubectl delete namespace kafka || echo "skipping"
-	kubectl delete -f https://github.com/jetstack/cert-manager/releases/download/v1.8.2/cert-manager.yaml || echo "skipping"
 
 # Deploys Venice cluster in docker and creates two stores in Venice. Stores are not managed via K8s for now.
 deploy-venice: deploy deploy-flink
-	docker compose -f ./deploy/docker/docker-compose-single-dc-setup.yaml up -d --wait
+	docker compose -f ./deploy/docker/venice/docker-compose-single-dc-setup.yaml up -d --wait
 	docker exec venice-client ./create-store.sh http://venice-controller:5555 venice-cluster0 test-store schemas/keySchema.avsc schemas/valueSchema.avsc
 	docker exec venice-client ./create-store.sh http://venice-controller:5555 venice-cluster0 test-store-1 schemas/keySchema.avsc schemas/valueSchema.avsc
 	kubectl apply -f ./deploy/samples/venicedb.yaml
 
 undeploy-venice:
 	kubectl delete -f ./deploy/samples/venicedb.yaml || echo "skipping"
-	docker compose -f ./deploy/docker/docker-compose-single-dc-setup.yaml down
+	docker compose -f ./deploy/docker/venice/docker-compose-single-dc-setup.yaml down
 
 deploy-dev-environment: deploy deploy-flink deploy-kafka deploy-venice
 

README.md

@@ -48,9 +48,9 @@ The below setup will install two local demo DBs, ads and profiles.
   > !intro
 ```
 
-## Set up Kafka & Flink clusters
+## Set up dev environment
 
-The below setup will install a Kafka and Flink cluster within Kubernetes.
+The below setup will create a dev environment with various resources within Kubernetes.
 
 ```
   $ make install                                                    # build and install SQL CLI
@@ -60,6 +60,26 @@ The below setup will install a Kafka and Flink cluster within Kubernetes.
   > !intro
 ```
 
+Commands `deploy-kafka`, `deploy-venice`, `deploy-flink`, etc. exist in isolation to deploy individual components.
+
+### Flink
+
+```
+  $ kubectl get pods
+  NAME                                              READY   STATUS    RESTARTS      AGE
+  basic-session-deployment-7b94b98b6b-d6jt5         1/1     Running   0             43s
+```
+
+Once the Flink deployment pod has STATUS 'Running', you can forward port 8081 and connect to http://localhost:8081/
+to access the Flink dashboard.
+
+```
+  $ kubectl port-forward basic-session-deployment-7b94b98b6b-d6jt5 8081 &
+```
+
+See the [Flink SQL Gateway Documentation](https://nightlies.apache.org/flink/flink-docs-release-1.18/docs/dev/table/sql-gateway/overview/)
+for sample adhoc queries through Flink.
+
 ## The SQL CLI
 
 The `./hoptimator` script launches the [sqlline](https://github.com/julianhyde/sqlline) SQL CLI pre-configured to connect to `jdbc:hoptimator://`. The CLI includes some additional commands. See `!intro`.

deploy/docker/flink/docker-compose-sql-gateway.yaml

@@ -0,0 +1,12 @@
+services:
+  flink-sql-gateway:
+    image: flink:1.18.1
+    restart: unless-stopped
+    entrypoint: >
+      /bin/sh -c "./bin/sql-gateway.sh start-foreground -Dsql-gateway.endpoint.rest.address=localhost"
+    ports:
+      - 8083:8083
+    deploy:
+      resources:
+        limits:
+          memory: 1024M

deploy/docker/docker-compose-single-dc-setup.yaml renamed to deploy/docker/venice/docker-compose-single-dc-setup.yaml

@@ -78,7 +78,7 @@ services:
     hostname: venice-client
     tty: true
     volumes:
-      - ./venice:/opt/venice/schemas
+      - ./schemas:/opt/venice/schemas
     depends_on:
       venice-router:
         condition: service_healthy

deploy/docker/venice/keySchema.avsc renamed to deploy/docker/venice/schemas/keySchema.avsc

@@ -0,0 +1,18 @@
+apiVersion: flink.apache.org/v1beta1
+kind: FlinkDeployment
+metadata:
+  name: basic-session-deployment
+spec:
+  image: flink:1.18
+  flinkVersion: v1_18
+  flinkConfiguration:
+    taskmanager.numberOfTaskSlots: "1"
+  serviceAccount: flink
+  jobManager:
+    resource:
+      memory: "2048m"
+      cpu: 0.1
+  taskManager:
+    resource:
+      memory: "2048m"
+      cpu: 0.1

deploy/docker/venice/valueSchema.avsc renamed to deploy/docker/venice/schemas/valueSchema.avsc

@@ -0,0 +1,22 @@
+## This template adds Flink support.
+
+apiVersion: hoptimator.linkedin.com/v1alpha1
+kind: JobTemplate
+metadata:
+  name: flink-template
+spec:
+  yaml: |
+    apiVersion: flink.apache.org/v1beta1
+    kind: FlinkSessionJob
+    metadata:
+      name: {{name}}
+    spec:
+      deploymentName: basic-session-deployment
+      job:
+        entryClass: com.linkedin.hoptimator.flink.runner.FlinkRunner
+        args:
+        - {{flinksql}}
+        jarURI: file:///opt/hoptimator-flink-runner.jar
+        parallelism: 1
+        upgradeMode: stateless
+        state: running