[WIP] Test Log forward

images/charmed-spark/bin/sparkd.sh

@@ -1,5 +1,15 @@
 #!/bin/bash
 
+function get_log_layer {
+  LOG_LAYER_FILE="/opt/pebble/log-layer.yaml"
+  ESCAPED_LOKI_URL="$(<<< "$LOKI_URL" sed -e 's`[][\\/.*^$]`\\&`g')"
+  sed -e "s/\$LOKI_URL/$ESCAPED_LOKI_URL/g" \
+      -e "s/\$SPARK_APPLICATION_ID/$SPARK_APPLICATION_ID/g" \
+      -e "s/\$SPARK_USER/$SPARK_USER/g" \
+      -e "s/\$SPARK_EXECUTOR_POD_NAME/$SPARK_EXECUTOR_POD_NAME/g" \
+      $LOG_LAYER_FILE
+}
+
 function finish {
   if [ $? -ne 0 ]
   then
@@ -9,6 +19,15 @@ function finish {
 }
 trap finish EXIT
 
+if [ ! -z "${LOKI_URL}" ]
+then
+    echo "Configuring log-forwarding to Loki."
+    RENDERED_LOG_LAYER=$(get_log_layer)
+    echo "$RENDERED_LOG_LAYER" | tee /tmp/rendered_log_layer.yaml
+    pebble add logging /tmp/rendered_log_layer.yaml
+else
+    echo "Log-forwarding to Loki is disabled."
+fi
 
 FLAVOUR=$1
 

images/charmed-spark/pebble/log-layer.yaml

@@ -0,0 +1,12 @@
+log-targets:
+  grafana-agent-k8s:
+    override: replace
+    type: loki
+    location: $LOKI_URL
+    services: [all]
+    labels:
+      product: charmed-spark
+      app: spark
+      app_id: $SPARK_APPLICATION_ID
+      user: $SPARK_USER
+      pod: $SPARK_EXECUTOR_POD_NAME

images/charmed-spark/rockcraft.yaml

@@ -182,6 +182,7 @@ parts:
       bin/spark-client.service-account-registry: opt/spark-client/python/bin/spark-client.service-account-registry
       bin/spark-client.spark-shell: opt/spark-client/python/bin/spark-client.spark-shell
       bin/spark-client.spark-submit: opt/spark-client/python/bin/spark-client.spark-submit
+      pebble/log-layer.yaml: opt/pebble/log-layer.yaml
     stage:
       - etc/spark8t/conf/
       - etc/spark/conf/
@@ -192,6 +193,7 @@ parts:
       - opt/spark-client/python/bin/spark-client.service-account-registry
       - opt/spark-client/python/bin/spark-client.spark-shell
       - opt/spark-client/python/bin/spark-client.spark-submit
+      - opt/pebble/log-layer.yaml
 
   user-setup:
     plugin: nil

tests/integration/integration-tests.sh

@@ -53,6 +53,19 @@ validate_metrics() {
   fi
 }
 
+validate_logs() {
+  log=$1
+  echo "LOG: "
+  cat $log
+  echo "END of LOG"
+  if [ $(grep -Ri "Configuring log-forwarding to Loki." $log | wc -l) -lt 2 ]; then
+    exit 1
+  fi
+  if [ $(grep -Ri 'Layer \\\\"logging\\\\" added successfully from \\\\"/tmp/rendered_log_layer.yaml\\\\"' $log | wc -l) -lt 2 ]; then
+    exit 1
+  fi
+}
+
 setup_user() {
   echo "setup_user() ${1} ${2}"
 
@@ -422,6 +435,56 @@ run_example_job_in_pod_with_metrics() {
 }
 
 
+run_example_job_in_pod_with_log_forwarding() {
+  NAMESPACE=${1-$NAMESPACE}
+  USERNAME=${2-spark}
+  SPARK_EXAMPLES_JAR_NAME="spark-examples_2.12-$(get_spark_version).jar"
+
+  PREVIOUS_JOB=$(kubectl -n $NAMESPACE get pods --sort-by=.metadata.creationTimestamp | grep driver | tail -n 1 | cut -d' ' -f1)
+  # start simple http server
+  LOG_FILE="/tmp/server_log_forward.log"
+  SERVER_PORT=9091
+  python3 tests/integration/resources/test_web_server.py $SERVER_PORT > $LOG_FILE &
+  HTTP_SERVER_PID=$!
+  # get ip address
+  IP_ADDRESS=$(hostname -I | cut -d ' ' -f 1)
+  echo "IP: $IP_ADDRESS"
+
+  kubectl -n $NAMESPACE exec testpod -- env PORT="$SERVER_PORT" IP="$IP_ADDRESS" UU="$USERNAME" NN="$NAMESPACE" JJ="$SPARK_EXAMPLES_JAR_NAME" IM="$(spark_image)" \
+                  /bin/bash -c 'spark-client.spark-submit -v \
+                  --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.executorEnv.LOKI_URL="$IP:$PORT" \
+                  --conf spark.driverEnv.LOKI_URL="$IP:$PORT" \
+                  --class org.apache.spark.examples.SparkPi \
+                  local:///opt/spark/examples/jars/$JJ 1000'
+
+  # kubectl --kubeconfig=${KUBE_CONFIG} get pods
+  DRIVER_PODS=$(kubectl get pods --sort-by=.metadata.creationTimestamp -n ${NAMESPACE} | grep driver )
+  DRIVER_JOB=$(kubectl get pods --sort-by=.metadata.creationTimestamp -n ${NAMESPACE} | grep driver | tail -n 1 | cut -d' ' -f1)
+
+  if [[ "${DRIVER_JOB}" == "${PREVIOUS_JOB}" ]]
+  then
+    echo "ERROR: Sample job has not run!"
+    exit 1
+  fi
+
+  # Check job output
+  # Sample output
+  # "Pi is roughly 3.13956232343"
+  pi=$(kubectl logs $(kubectl get pods --sort-by=.metadata.creationTimestamp -n ${NAMESPACE} | grep driver | tail -n 1 | cut -d' ' -f1)  -n ${NAMESPACE} | grep 'Pi is roughly' | rev | cut -d' ' -f1 | rev | cut -c 1-3)
+  echo -e "Spark Pi Job Output: \n ${pi}"
+
+  validate_pi_value $pi
+  validate_logs $LOG_FILE
+
+  # kill http server
+  kill $HTTP_SERVER_PID
+}
+
+
 run_example_job_with_error_in_pod() {
   SPARK_EXAMPLES_JAR_NAME="spark-examples_2.12-$(get_spark_version).jar"
 
@@ -657,6 +720,12 @@ echo -e "########################################"
 
 (setup_user_context && test_example_job_in_pod_with_metrics && cleanup_user_success) || cleanup_user_failure_in_pod
 
+echo -e "########################################"
+echo -e "RUN EXAMPLE JOB WITH LOG FORWARDING"
+echo -e "########################################"
+
+(setup_user_context && run_example_job_in_pod_with_log_forwarding && cleanup_user_success) || cleanup_user_failure_in_pod
+
 echo -e "########################################"
 echo -e "RUN EXAMPLE JOB WITH ERRORS"
 echo -e "########################################"