[DPE-3196] Add integration test for Apache Iceberg integration

.github/workflows/build.yaml

@@ -44,6 +44,9 @@ jobs:
           ARTIFACT=$(make help | grep 'Artifact: ')
           echo "name=${ARTIFACT#'Artifact: '}" >> $GITHUB_OUTPUT
 
+      - name: Change artifact permissions
+        run: sudo chmod a+r ${{ steps.artifact.outputs.name }}
+
       - name: Upload locally built artifact
         uses: actions/upload-artifact@v3
         with:

.gitignore

@@ -1 +1,4 @@
-.idea
+.idea
+*.rock
+*.tar
+.make_cache

CONTRIBUTING.md

@@ -13,6 +13,7 @@ cd charmed-spark-rock
 sudo snap install rockcraft --edge
 sudo snap install docker
 sudo snap install lxd
+sudo snap install yq
 sudo snap install skopeo --edge --devmode
 ```
 

Makefile

@@ -24,6 +24,7 @@ _MAKE_DIR := .make_cache
 $(shell mkdir -p $(_MAKE_DIR))
 
 K8S_TAG := $(_MAKE_DIR)/.k8s_tag
+AWS_TAG := $(_MAKE_DIR)/.aws_tag
 
 IMAGE_NAME := $(shell yq .name rockcraft.yaml)
 VERSION := $(shell yq .version rockcraft.yaml)
@@ -70,7 +71,7 @@ $(_TMP_OCI_TAG): $(_ROCK_OCI)
 	touch $(_TMP_OCI_TAG)
 
 $(CHARMED_OCI_TAG): $(_TMP_OCI_TAG)
-	docker build - -t "$(CHARMED_OCI_FULL_NAME):$(TAG)" --build-arg BASE_IMAGE="$(_TMP_OCI_NAME):$(TAG)" < Dockerfile
+	docker build -t "$(CHARMED_OCI_FULL_NAME):$(TAG)" --build-arg BASE_IMAGE="$(_TMP_OCI_NAME):$(TAG)" -f Dockerfile .
 	if [ ! -d "$(_MAKE_DIR)/$(CHARMED_OCI_FULL_NAME)" ]; then mkdir -p "$(_MAKE_DIR)/$(CHARMED_OCI_FULL_NAME)"; fi
 	touch $(CHARMED_OCI_TAG)
 
@@ -80,10 +81,15 @@ $(K8S_TAG):
 	sg microk8s ./tests/integration/config-microk8s.sh
 	@touch $(K8S_TAG)
 
+$(AWS_TAG): $(K8S_TAG)
+	@echo "=== Setting up and configure AWS CLI ==="
+	/bin/bash ./tests/integration/setup-aws-cli.sh
+	touch $(AWS_TAG)
+
 microk8s: $(K8S_TAG)
 
 $(_MAKE_DIR)/%/$(TAG).tar: $(_MAKE_DIR)/%/$(TAG).tag
-	docker save $*:$(TAG) > $(_MAKE_DIR)/$*/$(TAG).tar
+	docker save $*:$(TAG) -o $(_MAKE_DIR)/$*/$(TAG).tar
 
 $(BASE_NAME): $(_MAKE_DIR)/$(CHARMED_OCI_FULL_NAME)/$(TAG).tar
 	@echo "=== Creating $(BASE_NAME) OCI archive ==="
@@ -106,7 +112,7 @@ import: $(K8S_TAG) build
 	microk8s ctr images import --base-name $(CHARMED_OCI_FULL_NAME):$(TAG) $(BASE_NAME)
 endif
 
-tests:
+tests: $(K8S_TAG) $(AWS_TAG)
 	@echo "=== Running Integration Tests ==="
 	/bin/bash ./tests/integration/integration-tests.sh
 

tests/integration/config-microk8s.sh

@@ -2,3 +2,4 @@ microk8s status --wait-ready
 microk8s config | tee ~/.kube/config
 microk8s.enable dns
 microk8s.enable rbac
+microk8s.enable minio

tests/integration/integration-tests.sh

@@ -135,7 +135,7 @@ teardown_test_pod() {
 run_example_job_in_pod() {
   SPARK_EXAMPLES_JAR_NAME="spark-examples_2.12-$(get_spark_version).jar"
 
-  PREVIOUS_JOB=$(kubectl get pods | grep driver | tail -n 1 | cut -d' ' -f1)
+  PREVIOUS_JOB=$(kubectl get pods -n ${NAMESPACE}| grep driver | tail -n 1 | cut -d' ' -f1)
   NAMESPACE=$1
   USERNAME=$2
 
@@ -166,6 +166,101 @@ run_example_job_in_pod() {
   validate_pi_value $pi
 }
 
+get_s3_access_key(){
+  kubectl get secret -n minio-operator microk8s-user-1 -o jsonpath='{.data.CONSOLE_ACCESS_KEY}' | base64 -d
+}
+
+get_s3_secret_key(){
+  kubectl get secret -n minio-operator microk8s-user-1 -o jsonpath='{.data.CONSOLE_SECRET_KEY}' | base64 -d
+}
+
+get_s3_endpoint(){
+  kubectl get service minio -n minio-operator -o jsonpath='{.spec.clusterIP}'
+}
+
+create_s3_bucket(){
+  S3_ENDPOINT=$(get_s3_endpoint)
+  BUCKET_NAME=$1
+  aws --endpoint-url "http://$S3_ENDPOINT" s3api create-bucket --bucket "$BUCKET_NAME"
+  echo "Created S3 bucket ${BUCKET_NAME}"
+}
+
+delete_s3_bucket(){
+  S3_ENDPOINT=$(get_s3_endpoint)
+  BUCKET_NAME=$1
+  aws --endpoint-url "http://$S3_ENDPOINT" s3 rb "s3://$BUCKET_NAME" --force
+  echo "Deleted S3 bucket ${BUCKET_NAME}"
+}
+
+copy_file_to_s3_bucket(){
+  BUCKET_NAME=$1
+  FILE_PATH=$2
+  BASE_NAME=$(basename "$FILE_PATH")
+  S3_ENDPOINT=$(get_s3_endpoint)
+  aws --endpoint-url "http://$S3_ENDPOINT" s3 cp $FILE_PATH s3://"$BUCKET_NAME"/"$BASE_NAME"
+  echo "Copied file ${FILE_PATH} to S3 bucket ${BUCKET_NAME}"
+}
+
+test_iceberg_example_in_pod(){
+  create_s3_bucket spark
+  copy_file_to_s3_bucket spark ./tests/integration/resources/test-iceberg.py
+
+  NAMESPACE="tests"
+  USERNAME="spark"
+  NUM_ROWS_TO_INSERT="4"
+  PREVIOUS_DRIVER_PODS_COUNT=$(kubectl get pods -n ${NAMESPACE} | grep driver | wc -l)
+
+  kubectl exec testpod -- \
+      env \
+        UU="$USERNAME" \
+        NN="$NAMESPACE" \
+        IM="$(spark_image)" \
+        NUM_ROWS="$NUM_ROWS_TO_INSERT" \
+        ACCESS_KEY="$(get_s3_access_key)" \
+        SECRET_KEY="$(get_s3_secret_key)" \
+        S3_ENDPOINT="$(get_s3_endpoint)" \
+      /bin/bash -c '\
+        spark-client.spark-submit \
+        --username $UU --namespace $NN \
+        --conf spark.kubernetes.driver.request.cores=100m \
+        --conf spark.kubernetes.executor.request.cores=100m \
+        --conf spark.kubernetes.container.image=$IM \
+        --conf spark.hadoop.fs.s3a.aws.credentials.provider=org.apache.hadoop.fs.s3a.SimpleAWSCredentialsProvider \
+        --conf spark.hadoop.fs.s3a.connection.ssl.enabled=false \
+        --conf spark.hadoop.fs.s3a.path.style.access=true \
+        --conf spark.hadoop.fs.s3a.endpoint=$S3_ENDPOINT \
+        --conf spark.hadoop.fs.s3a.access.key=$ACCESS_KEY \
+        --conf spark.hadoop.fs.s3a.secret.key=$SECRET_KEY \
+        --conf spark.jars.ivy=/tmp \
+        --conf spark.sql.extensions=org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions \
+        --conf spark.sql.catalog.spark_catalog=org.apache.iceberg.spark.SparkSessionCatalog \
+        --conf spark.sql.catalog.spark_catalog.type=hive \
+        --conf spark.sql.catalog.local=org.apache.iceberg.spark.SparkCatalog \
+        --conf spark.sql.catalog.local.type=hadoop \
+        --conf spark.sql.catalog.local.warehouse=s3a://spark/warehouse \
+        --conf spark.sql.defaultCatalog=local \
+        s3a://spark/test-iceberg.py -n $NUM_ROWS'
+
+  delete_s3_bucket spark
+  DRIVER_PODS_COUNT=$(kubectl get pods -n ${NAMESPACE} | grep driver | wc -l)
+
+  if [[ "${PREVIOUS_DRIVER_PODS_COUNT}" == "${DRIVER_PODS_COUNT}" ]]
+  then
+    echo "ERROR: Sample job has not run!"
+    exit 1
+  fi
+
+  DRIVER_POD_ID=$(kubectl get pods -n ${NAMESPACE} | grep test-iceberg-.*-driver | tail -n 1 | cut -d' ' -f1)
+  OUTPUT_LOG_LINE=$(kubectl logs ${DRIVER_POD_ID} -n ${NAMESPACE} | grep 'Number of rows inserted:' )
+  NUM_ROWS_INSERTED=$(echo $OUTPUT_LOG_LINE | rev | cut -d' ' -f1 | rev)
+
+  if [ "${NUM_ROWS_INSERTED}" != "${NUM_ROWS_TO_INSERT}" ]; then
+      echo "ERROR: ${NUM_ROWS_TO_INSERT} were supposed to be inserted. Found ${NUM_ROWS_INSERTED} rows. Aborting with exit code 1."
+      exit 1
+  fi
+
+}
+
 run_example_job_in_pod_with_pod_templates() {
   SPARK_EXAMPLES_JAR_NAME="spark-examples_2.12-$(get_spark_version).jar"
 
@@ -425,6 +520,13 @@ echo -e "RUN EXAMPLE JOB WITH ERRORS"
 echo -e "########################################"
 
 (setup_user_admin_context && test_example_job_in_pod_with_errors && cleanup_user_success) || cleanup_user_failure_in_pod
+
+echo -e "##################################"
+echo -e "RUN EXAMPLE THAT USES ICEBERG LIBRARIES"
+echo -e "##################################"
+
+(setup_user_admin_context && test_iceberg_example_in_pod && cleanup_user_success) || cleanup_user_failure_in_pod
+
 echo -e "##################################"
 echo -e "TEARDOWN TEST POD"
 echo -e "##################################"

tests/integration/resources/test-iceberg.py

@@ -0,0 +1,30 @@
+import argparse
+import random
+
+from pyspark.sql import SparkSession
+from pyspark.sql.types import LongType, StructType, StructField
+
+parser = argparse.ArgumentParser("TestIceberg")
+parser.add_argument("--num_rows", "-n", type=int)
+args = parser.parse_args()
+num_rows = args.num_rows
+
+spark = SparkSession.builder.appName("IcebergExample").getOrCreate()
+
+
+schema = StructType(
+    [StructField("row_id", LongType(), True), StructField("row_val", LongType(), True)]
+)
+
+data = []
+for idx in range(num_rows):
+    row = (idx + 1, random.randint(1, 100))
+    data.append(row)
+
+df = spark.createDataFrame(data, schema)
+df.writeTo("demo.foo.bar").create()
+
+
+df = spark.table("demo.foo.bar")
+count = df.count()
+print(f"Number of rows inserted: {count}")

tests/integration/setup-aws-cli.sh

@@ -0,0 +1,17 @@
+#!/bin/bash
+
+# Install AWS CLI
+sudo snap install aws-cli --classic
+
+# Get Access key and secret key from MinIO
+ACCESS_KEY=$(kubectl get secret -n minio-operator microk8s-user-1 -o jsonpath='{.data.CONSOLE_ACCESS_KEY}' | base64 -d)
+SECRET_KEY=$(kubectl get secret -n minio-operator microk8s-user-1 -o jsonpath='{.data.CONSOLE_SECRET_KEY}' | base64 -d)
+
+S3_BUCKET="spark"
+DEFAULT_REGION="us-east-2"
+
+# Configure AWS CLI credentials
+aws configure set aws_access_key_id $ACCESS_KEY
+aws configure set aws_secret_access_key $SECRET_KEY
+aws configure set default.region $DEFAULT_REGION
+echo "AWS CLI credentials set successfully"