tests: spark connect/kerberos/iceberg integration test

docs/modules/spark-k8s/pages/usage-guide/spark-connect.adoc

@@ -97,6 +97,92 @@ spec:
 ...
 ```
 
+== Kerberos
+
+NOTE: Kerberos support for Spark Connect is not a first-class feature of the operator. The setup described here is a manual configuration that uses `podOverrides`, `configOverrides`, `envOverrides` and an init container.
+
+A Spark Connect server can authenticate to a Kerberos-secured service, such as a Apache Hive metastore or Apache Hadoop HDFS.
+There is, however, an important caveat: the Spark Connect server runs in Spark's `client` deploy mode.
+Spark only performs the automatic keytab login (`UserGroupInformation.loginUserFromKeytab`) for YARN, local, Mesos and the Kubernetes _cluster_-mode driver -- *not* for the Kubernetes client mode that the Connect server uses.
+As a consequence, setting `spark.kerberos.keytab` and `spark.kerberos.principal` alone does not obtain a Kerberos ticket (TGT), and the SASL/GSSAPI handshake to the metastore fails with `GSS initiate failed` / `Failed to find any Kerberos tgt`.
+
+The workaround is to obtain the ticket yourself with an *init container* that runs `kinit` before the Spark Connect server JVM starts.
+The init container reads the keytab and writes a Kerberos credential cache to an `emptyDir` volume that is shared with the `spark` container; the server JVM then picks the ticket up from that cache.
+An init container is required because nothing in the client-mode server process performs the login automatically, so the ticket must exist in the credential cache before the JVM opens the metastore connection.
+
+The relevant parts of the `SparkConnectServer` are shown below.
+
+[source,yaml]
+----
+spec:
+  server:
+    envOverrides:
+      # The shared credential cache populated by the init container. Hadoop's
+      # UserGroupInformation reads KRB5CCNAME when Kerberos auth is enabled.
+      KRB5CCNAME: /stackable/krb5/ccache
+    jvmArgumentOverrides:
+      add:
+        # The JVM does NOT read the KRB5_CONFIG environment variable (that is an
+        # MIT C-library variable). It reads this system property (or /etc/krb5.conf).
+        - -Djava.security.krb5.conf=/stackable/kerberos/krb5.conf
+        # The Spark Connect execute thread does not carry the ticket in its
+        # Subject, so JGSS must fall back to the ambient credential cache.
+        - -Djavax.security.auth.useSubjectCredsOnly=false
+    configOverrides:
+      spark-defaults.conf:
+        spark.hadoop.hadoop.security.authentication: "kerberos"
+        spark.hadoop.hive.metastore.uris: "thrift://hive-metastore:9083"
+        spark.hadoop.hive.metastore.sasl.enabled: "true"
+        # Use the literal metastore principal, not _HOST (which would resolve to
+        # the connection host instead of the metastore's service principal).
+        spark.hadoop.hive.metastore.kerberos.principal: "hive/hive.example.svc.cluster.local@EXAMPLE.COM"
+    podOverrides:
+      spec:
+        initContainers:
+          - name: kinit
+            image: oci.stackable.tech/sdp/spark-k8s:4.1.1-stackable0.0.0-dev
+            command: ["/bin/bash", "-euo", "pipefail", "-c"]
+            args:
+              - kinit -kt /stackable/kerberos/keytab spark-connect/spark-connect.example.svc.cluster.local@EXAMPLE.COM
+            env:
+              - name: KRB5_CONFIG
+                value: /stackable/kerberos/krb5.conf
+              - name: KRB5CCNAME
+                value: /stackable/krb5/ccache
+            volumeMounts:
+              - name: kerberos
+                mountPath: /stackable/kerberos
+              - name: krb5-ccache
+                mountPath: /stackable/krb5
+        containers:
+          - name: spark
+            volumeMounts:
+              - name: kerberos
+                mountPath: /stackable/kerberos
+              - name: krb5-ccache
+                mountPath: /stackable/krb5
+        volumes:
+          - name: krb5-ccache
+            emptyDir: {}
+          - name: kerberos
+            ephemeral:
+              volumeClaimTemplate:
+                metadata:
+                  annotations:
+                    secrets.stackable.tech/class: kerberos
+                    secrets.stackable.tech/scope: service=spark-connect
+                    secrets.stackable.tech/kerberos.service.names: spark-connect
+                spec:
+                  storageClassName: secrets.stackable.tech
+                  accessModes: ["ReadWriteOnce"]
+                  resources:
+                    requests:
+                      storage: "1"
+----
+
+The keytab and `krb5.conf` are only required on the server, which is the Spark driver and the only component that talks to the metastore.
+Executors that merely read and write data files on (non-kerberized) S3 do not need Kerberos credentials.
+
 == Spark History Server
 
 Unfortunately integration with the Spark History Server is not supported yet.
@@ -112,4 +198,9 @@ The following features are not supported by the Stackable Spark operator yet
 
 == Known Issues
 
-* Dynamically provisioning the iceberg runtime leads to "iceberg.SparkWrite$WriterFactory" ClassNotfoundException when attempting to use it from clients.
+* Distributed operations on Apache Iceberg tables fail on the executors *with Spark 3.5.x*.
+  PySpark calls that trigger a distributed job over an Iceberg table -- for example `DataFrame.collect()` or `DataFrame.count()` -- fail with `java.lang.ClassCastException: cannot assign instance of java.lang.invoke.SerializedLambda to field org.apache.spark.rdd.MapPartitionsRDD.f of type scala.Function3`.
+  Driver-only operations still work: DDL (`CREATE`/`DROP`), `INSERT` of small data, `DataFrame.show()` of a small result, `SHOW TABLES` and `DESCRIBE`.
+  The cause is upstream in Spark Connect: executor tasks run under a per-session class loader (which fetches session classes from the driver's artifact server) that cannot deserialize the Iceberg scan closures.
+  It is independent of how the Iceberg runtime is provisioned -- it reproduces with `--packages`, with the jar placed on the system class path (`/stackable/spark/jars`), with `spark.addArtifact()`, and with combinations of these.
+  *This is fixed in Spark 4* (https://issues.apache.org/jira/browse/SPARK-51537[SPARK-51537] / https://github.com/apache/spark/pull/50475[apache/spark#50475]), where distributed reads and writes of Iceberg tables work; it is not backported to the 3.5.x line (see the still-open https://issues.apache.org/jira/browse/SPARK-46032[SPARK-46032]).

tests/templates/kuttl/spark-connect-kerberos/00-assert.yaml

@@ -0,0 +1,9 @@
+---
+apiVersion: kuttl.dev/v1beta1
+kind: TestAssert
+timeout: 900
+---
+apiVersion: v1
+kind: ServiceAccount
+metadata:
+  name: integration-tests-sa

tests/templates/kuttl/spark-connect-kerberos/00-patch-ns.yaml.j2

@@ -0,0 +1,9 @@
+{% if test_scenario['values']['openshift'] == 'true' %}
+# see https://github.com/stackabletech/issues/issues/566
+---
+apiVersion: kuttl.dev/v1beta1
+kind: TestStep
+commands:
+  - script: kubectl patch namespace $NAMESPACE -p '{"metadata":{"labels":{"pod-security.kubernetes.io/enforce":"privileged"}}}'
+    timeout: 120
+{% endif %}

tests/templates/kuttl/spark-connect-kerberos/00-serviceaccount.yaml.j2

@@ -0,0 +1,29 @@
+---
+kind: Role
+apiVersion: rbac.authorization.k8s.io/v1
+metadata:
+  name: use-integration-tests-scc
+rules:
+{% if test_scenario['values']['openshift'] == "true" %}
+  - apiGroups: ["security.openshift.io"]
+    resources: ["securitycontextconstraints"]
+    resourceNames: ["privileged"]
+    verbs: ["use"]
+{% endif %}
+---
+apiVersion: v1
+kind: ServiceAccount
+metadata:
+  name: integration-tests-sa
+---
+kind: RoleBinding
+apiVersion: rbac.authorization.k8s.io/v1
+metadata:
+  name: use-integration-tests-scc
+subjects:
+  - kind: ServiceAccount
+    name: integration-tests-sa
+roleRef:
+  kind: Role
+  name: use-integration-tests-scc
+  apiGroup: rbac.authorization.k8s.io

tests/templates/kuttl/spark-connect-kerberos/01-assert.yaml.j2

@@ -0,0 +1,10 @@
+---
+apiVersion: kuttl.dev/v1beta1
+kind: TestAssert
+{% if lookup('env', 'VECTOR_AGGREGATOR') %}
+---
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: vector-aggregator-discovery
+{% endif %}

tests/templates/kuttl/spark-connect-kerberos/01-install-vector-aggregator-discovery-configmap.yaml.j2

@@ -0,0 +1,9 @@
+{% if lookup('env', 'VECTOR_AGGREGATOR') %}
+---
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: vector-aggregator-discovery
+data:
+  ADDRESS: {{ lookup('env', 'VECTOR_AGGREGATOR') }}
+{% endif %}

tests/templates/kuttl/spark-connect-kerberos/02-assert.yaml

@@ -0,0 +1,12 @@
+---
+apiVersion: kuttl.dev/v1beta1
+kind: TestAssert
+timeout: 600
+---
+apiVersion: apps/v1
+kind: StatefulSet
+metadata:
+  name: krb5-kdc
+status:
+  readyReplicas: 1
+  replicas: 1

tests/templates/kuttl/spark-connect-kerberos/02-install-krb5-kdc.yaml.j2

@@ -0,0 +1,142 @@
+---
+apiVersion: apps/v1
+kind: StatefulSet
+metadata:
+  name: krb5-kdc
+spec:
+  selector:
+    matchLabels:
+      app: krb5-kdc
+  template:
+    metadata:
+      labels:
+        app: krb5-kdc
+    spec:
+      serviceAccountName: integration-tests-sa
+      initContainers:
+        - name: init
+          image: oci.stackable.tech/sdp/krb5:{{ test_scenario['values']['krb5'] }}-stackable0.0.0-dev
+          args:
+            - sh
+            - -euo
+            - pipefail
+            - -c
+            - |
+              test -e /var/kerberos/krb5kdc/principal || kdb5_util create -s -P asdf
+              kadmin.local get_principal -terse root/admin || kadmin.local add_principal -pw asdf root/admin
+              # stackable-secret-operator principal must match the keytab specified in the SecretClass
+              kadmin.local get_principal -terse stackable-secret-operator || kadmin.local add_principal -e aes256-cts-hmac-sha384-192:normal -pw asdf stackable-secret-operator
+          env:
+            - name: KRB5_CONFIG
+              value: /stackable/config/krb5.conf
+          volumeMounts:
+            - mountPath: /stackable/config
+              name: config
+            - mountPath: /var/kerberos/krb5kdc
+              name: data
+      containers:
+        - name: kdc
+          image: oci.stackable.tech/sdp/krb5:{{ test_scenario['values']['krb5'] }}-stackable0.0.0-dev
+          args:
+            - krb5kdc
+            - -n
+          env:
+            - name: KRB5_CONFIG
+              value: /stackable/config/krb5.conf
+          volumeMounts:
+            - mountPath: /stackable/config
+              name: config
+            - mountPath: /var/kerberos/krb5kdc
+              name: data
+# Root permissions required on Openshift to access internal ports
+{% if test_scenario['values']['openshift'] == "true" %}
+          securityContext:
+            runAsUser: 0
+{% endif %}
+        - name: kadmind
+          image: oci.stackable.tech/sdp/krb5:{{ test_scenario['values']['krb5'] }}-stackable0.0.0-dev
+          args:
+            - kadmind
+            - -nofork
+          env:
+            - name: KRB5_CONFIG
+              value: /stackable/config/krb5.conf
+          volumeMounts:
+            - mountPath: /stackable/config
+              name: config
+            - mountPath: /var/kerberos/krb5kdc
+              name: data
+# Root permissions required on Openshift to access internal ports
+{% if test_scenario['values']['openshift'] == "true" %}
+          securityContext:
+            runAsUser: 0
+{% endif %}
+        - name: client
+          image: oci.stackable.tech/sdp/krb5:{{ test_scenario['values']['krb5'] }}-stackable0.0.0-dev
+          tty: true
+          stdin: true
+          env:
+            - name: KRB5_CONFIG
+              value: /stackable/config/krb5.conf
+          volumeMounts:
+            - mountPath: /stackable/config
+              name: config
+      volumes:
+        - name: config
+          configMap:
+            name: krb5-kdc
+  volumeClaimTemplates:
+    - metadata:
+        name: data
+      spec:
+        accessModes:
+          - ReadWriteOnce
+        resources:
+          requests:
+            storage: 1Gi
+---
+apiVersion: v1
+kind: Service
+metadata:
+  name: krb5-kdc
+spec:
+  selector:
+    app: krb5-kdc
+  ports:
+    - name: kadmin
+      port: 749
+    - name: kdc
+      port: 88
+    - name: kdc-udp
+      port: 88
+      protocol: UDP
+---
+apiVersion: v1
+kind: ConfigMap
+metadata:
+  name: krb5-kdc
+data:
+  krb5.conf: |
+    [logging]
+    default = STDERR
+    kdc = STDERR
+    admin_server = STDERR
+    [libdefaults]
+    dns_lookup_realm = false
+    ticket_lifetime = 24h
+    renew_lifetime = 7d
+    forwardable = true
+    rdns = false
+    default_realm = {{ test_scenario['values']['kerberos-realm'] }}
+    spake_preauth_groups = edwards25519
+    [realms]
+    {{ test_scenario['values']['kerberos-realm'] }} = {
+     acl_file = /stackable/config/kadm5.acl
+     disable_encrypted_timestamp = false
+    }
+    [domain_realm]
+    .cluster.local = {{ test_scenario['values']['kerberos-realm'] }}
+    cluster.local = {{ test_scenario['values']['kerberos-realm'] }}
+  kadm5.acl: |
+    root/admin *e
+    stackable-secret-operator *e

tests/templates/kuttl/spark-connect-kerberos/03-create-kerberos-secretclass.yaml.j2

@@ -0,0 +1,42 @@
+---
+apiVersion: kuttl.dev/v1beta1
+kind: TestStep
+commands:
+  # We need to replace $NAMESPACE (by KUTTL)
+  - script: |
+      kubectl apply -n "$NAMESPACE" -f - <<EOF
+      ---
+      apiVersion: secrets.stackable.tech/v1alpha1
+      kind: SecretClass
+      metadata:
+        name: kerberos-$NAMESPACE
+      spec:
+        backend:
+          kerberosKeytab:
+            realmName: {{ test_scenario['values']['kerberos-realm'] }}
+            kdc: krb5-kdc.$NAMESPACE.svc.cluster.local
+            admin:
+              mit:
+                kadminServer: krb5-kdc.$NAMESPACE.svc.cluster.local
+            adminKeytabSecret:
+              namespace: $NAMESPACE
+              name: secret-operator-keytab
+            adminPrincipal: stackable-secret-operator
+      EOF
+---
+apiVersion: v1
+kind: Secret
+metadata:
+  name: secret-operator-keytab
+data:
+  # To create keytab. When prompted enter password asdf
+  # cat | ktutil << 'EOF'
+  # list
+  # add_entry -password -p stackable-secret-operator@CLUSTER.LOCAL -k 1 -e aes256-cts-hmac-sha384-192
+  # wkt /tmp/keytab
+  # EOF
+{% if test_scenario['values']['kerberos-realm'] == 'CLUSTER.LOCAL' %}
+  keytab: BQIAAABdAAEADUNMVVNURVIuTE9DQUwAGXN0YWNrYWJsZS1zZWNyZXQtb3BlcmF0b3IAAAABZAYWIgEAFAAgm8MCZ8B//XF1tH92GciD6/usWUNAmBTZnZQxLua2TkgAAAAB
+{% elif test_scenario['values']['kerberos-realm'] == 'PROD.MYCORP' %}
+  keytab: BQIAAABbAAEAC1BST0QuTVlDT1JQABlzdGFja2FibGUtc2VjcmV0LW9wZXJhdG9yAAAAAWQZa0EBABQAIC/EnFNejq/K5lX6tX+B3/tkI13TCzkPB7d2ggCIEzE8AAAAAQ==
+{% endif %}