diff --git a/analytics/flink/README.md b/analytics/flink/README.md new file mode 100644 index 000000000..7f265e644 --- /dev/null +++ b/analytics/flink/README.md @@ -0,0 +1,200 @@ +# Apache Flink on EKS Auto Mode + +An in-workshop [Apache Flink](https://flink.apache.org/) runtime that sits alongside the Spark labs — and, when deployed together, the Kafka and ClickHouse labs — on the **same** EKS Auto Mode cluster created by `analytics/terraform/spark-k8s-operator/`. No extra VPC, no separate cluster, no side install. + +Under the hood: the [Apache Flink Kubernetes Operator](https://github.com/apache/flink-kubernetes-operator) manages a `FlinkDeployment` running the built-in `StateMachineExample.jar` from the official `flink:1.20` image. One JobManager plus two TaskManagers, in Application mode, on a dedicated Karpenter On-Demand NodePool. + +## Architecture at a glance + +Three design decisions shape this lab. Understanding them up-front makes the rest of the manifests obvious. + +### 1. Flink runs on a dedicated NodePool, kept away from Spark, Kafka, and ClickHouse + +A Flink JobManager owns the execution graph and checkpoint metadata; a TaskManager owns per-key state for its assigned partitions. Losing either forces a job restart from the last checkpoint — cheap for a demo, painful for a production stream that's built up hours of state. Same reasoning as Kafka and ClickHouse: keep them off Spot, off shared pools. + +The workshop's Terraform creates a **dedicated Karpenter `NodePool`** for Flink (see `analytics/terraform/spark-k8s-operator/manifests/automode/nodepool-flink.yaml`) that runs **On-Demand only** and carries a `workload=flink:NoSchedule` taint. The `FlinkDeployment` in this folder carries the matching toleration and nodeSelector. Nothing else in the workshop cluster tolerates that taint, so nothing else lands on those nodes. + +The routing uses **three complementary pieces** on top of Karpenter's usual scheduling: + +| Direction | Mechanism | Effect | +|---|---|---| +| Non-Flink pods **off** the Flink pool | `workload=flink:NoSchedule` taint on the NodePool | Scheduler refuses to place them | +| Flink pods **allowed on** the pool | Matching toleration on the JM and TM pod templates | Scheduler accepts placement | +| Flink pods **routed to** the pool | `workload: flink` label on the pool + matching `nodeSelector` on the pod | Karpenter provisions from this pool, not from a higher-weighted general-purpose pool | + +The nodeSelector matters as much as the taint. Without it, when a JobManager becomes pending, Karpenter picks the highest-weight *feasible* NodePool — since `general-purpose` weight=50 and our Flink pool has no weight, Flink would land on general-purpose and the dedicated pool would sit empty. Taint + toleration alone doesn't route. + +### 2. Application mode, not Session mode + +Two ways an operator can run Flink jobs: + +- **Session mode** — one long-lived Flink cluster hosts many jobs. Legacy pattern, jobs share the JobManager, resource isolation is by slot rather than pod. +- **Application mode** — one Flink cluster per job. The FlinkDeployment CR = one JobManager + N TaskManagers, and the job runs until you delete the CR. Modern operator-native pattern. + +This lab uses Application mode: `flink-cluster.yaml` defines a `FlinkDeployment` with an inline `job` block pointing at the StateMachine JAR. Deleting the CR removes the whole cluster — no leftover session waiting for another job. + +### 3. State + checkpoints, not stateless + +Every long-running stream processor has to answer two questions: where does per-key state live, and what happens when a pod dies? + +- **State backend** — the sample uses the `hashmap` (in-memory) backend for simplicity. Fine for the ~15 KB state the state-machine keeps per key; graduate to `rocksdb` when state grows past what fits in the JVM heap. +- **Checkpoints** — periodic snapshots of the state. Written to `file:///tmp/flink-checkpoints` on the JM pod for the demo (so a restart loses recent progress), swap to `s3://` for real fault-tolerance. +- **Savepoints** — manually-triggered snapshots you can restart from. The path is set in `flinkConfiguration.state.savepoints.dir`. + +For an HA JobManager you'd add `spec.jobManager.replicas: 2` and a `kubernetesHAOptions` block. The workshop keeps things single-JM to keep the demo readable. + +## Prerequisites + +The Spark-on-EKS workshop cluster is up (`analytics/terraform/spark-k8s-operator/` deployed) and `kubectl` targets it: + +```sh +kubectl get nodes -o wide | head +kubectl get storageclass # expect: gp3 (default), flink-gp3, plus kafka-gp3 / clickhouse-gp3 if those labs are enabled +kubectl get ns cert-manager # cert-manager is a prerequisite for the Flink operator's admission webhooks +``` + +`cert-manager` ships with the workshop's Terraform addon stack, so unless you disabled it, this last check should just pass. + +## Files + +``` +analytics/flink/ +├── README.md +├── deploy-flink.sh # applies flink-cluster.yaml, waits for FlinkDeployment to reach LIFECYCLE STATE=STABLE +├── cleanup.sh # removes the FlinkDeployment and its PVCs (Terraform owns the operator + NodePool + StorageClass) +└── flink-cluster.yaml # FlinkDeployment CR — StateMachineExample, Application mode, parallelism 2 +``` + +The Flink Kubernetes Operator itself is installed exclusively by the workshop's Terraform (see `analytics/terraform/spark-k8s-operator/flink-operator.tf`). There is no bash-wrapper install script — helm is invoked from Terraform's `helm_release` resource, so operator version and lifecycle are managed alongside the rest of the workshop infrastructure. + +## Deploy + +### Confirm the operator is running + +The Flink Kubernetes Operator is installed by the workshop's Terraform when `enable_flink_lab = true` (the default). Confirm it's ready before you apply the FlinkDeployment: + +```sh +kubectl -n flink rollout status deploy/flink-kubernetes-operator +kubectl get storageclass flink-gp3 +``` + +Both should be present. If either is missing, re-apply Terraform with `enable_flink_lab = true` — the operator, StorageClass, and NodePool all come from that single toggle. + +### Apply the sample FlinkDeployment + +```sh +./deploy-flink.sh +``` + +The Flink operator turns `flink-cluster.yaml` into: + +- **1× JobManager Deployment** (`state-machine`), 1 CPU / 2 GiB — the control plane pod; accepts the JAR, plans the execution graph, spawns TaskManagers via Flink's native Kubernetes API +- **2× TaskManager pods** (`state-machine-taskmanager-1-{1,2}`), 1 CPU / 2 GiB each — the workers running the operators +- **REST/UI Service** (`state-machine-rest`) — port 8081, the Flink Web UI and REST API +- **Internal Service** (`state-machine`) — cluster-internal communication between JM and TMs + +Karpenter provisions 1 or 2 On-Demand instances — typically `m5a.2xlarge` or `r5a.2xlarge` under the shipped requirements — and bin-packs the three Flink pods onto them. First deploy takes ~3-6 minutes end-to-end while the JVM warms up. When the CR reports `LIFECYCLE STATE=STABLE`: + +```sh +kubectl get flinkdeployment -n flink +kubectl get pods -n flink -o wide +kubectl get nodeclaims -l karpenter.sh/nodepool=flink +``` + +Endpoints (in-cluster): + +- **REST + Web UI:** `http://state-machine-rest.flink.svc:8081` +- **JobManager RPC:** `state-machine.flink.svc:6123` (internal) + +## Verify + +Unlike a Spark job that reads a file and finishes, a Flink job is **long-running**. The proof-of-running lives in three places: the CR status, the JobManager REST API, and the checkpoint counter. + +**1. FlinkDeployment status:** + +```sh +kubectl -n flink get flinkdeployment state-machine +``` + +Expected: + +``` +NAME JOB STATUS LIFECYCLE STATE +state-machine RUNNING STABLE +``` + +**2. Port-forward the JobManager REST / Web UI:** + +```sh +kubectl -n flink port-forward svc/state-machine-rest 8081:8081 +``` + +Open [http://localhost:8081](http://localhost:8081) — Flink Web UI with jobs, tasks, backpressure, thread dumps, and checkpoint history. + +**3. Query the REST API for job overview and checkpoint counts:** + +```sh +curl -s http://localhost:8081/jobs/overview | python3 -m json.tool + +JID=$(kubectl -n flink get flinkdeployment state-machine -o jsonpath='{.status.jobStatus.jobId}') +curl -s "http://localhost:8081/jobs/$JID/checkpoints" | python3 -c " +import sys, json +d = json.load(sys.stdin) +print('completed:', d['counts']['completed']) +print('failed:', d['counts']['failed']) +" +``` + +Expected — all 4 tasks (Source Generator × 2 parallelism + Flat Map + Print Sink × 2) running, and the completed checkpoint count climbing every few seconds. + +**Do not** grep TaskManager stdout for state-machine events — the sample only prints on invalid transitions, which are probability-driven and sparse. Job health lives in the status + REST API, not in tail logs. That's the streaming pattern, not the batch pattern. + +## Cleanup + +```sh +./cleanup.sh +``` + +Removes the FlinkDeployment and any PVCs it created. The Flink operator, the `flink-gp3` StorageClass, the dedicated Flink NodePool, and the `flink` namespace stay in place — those are owned by Terraform and go away with `terraform destroy` when you tear down the workshop. + +Because the StorageClass uses `reclaimPolicy: Retain`, deleting the PVCs does not delete the underlying EBS volumes on its own — the volumes stay in an available state until Terraform cleans them up during `destroy`. That's the trade-off for making the workshop safe against accidental `kubectl delete pvc`. + +## Storage tiers + +**Flink's I/O demand is bursty.** Every checkpoint interval each TaskManager writes a state delta; when RocksDB is enabled, background compaction competes for the same volume. The sample job uses filesystem checkpoints on ephemeral `/tmp` (so a JM restart loses recent progress) — good enough for a demo, wrong for production. Two upgrade paths: + +| Approach | When it makes sense | Trade-off | +|---|---|---| +| **`file:///tmp/flink-checkpoints`** (default) | Workshop demos, functional testing | Data lost on JobManager pod restart — not fault-tolerant | +| **PVC on `flink-gp3` + RocksDB local dir** | Local state larger than heap; single-node fault tolerance if pod is rescheduled to the same node | AZ-scoped EBS ties pod placement | +| **`s3:///flink-checkpoints`** | Real fault tolerance — any pod can recover from any other pod's checkpoint | Adds S3 latency to the checkpoint path | + +To switch the sample to S3, edit `flink-cluster.yaml`: + +```yaml +flinkConfiguration: + state.backend.type: rocksdb + state.checkpoints.dir: s3:///flink-checkpoints + state.savepoints.dir: s3:///flink-savepoints +``` + +Then give the JM + TM pods an S3 IRSA role so they can write, and re-apply. + +## Sizing + +Rules of thumb for a Flink cluster, worth knowing before you touch the resource requests in `flink-cluster.yaml`: + +- **Heap sizing is a Flink concern, not just a JVM one.** Flink's memory model splits the container into JVM heap, JVM off-heap, managed memory (for RocksDB), network buffers, and framework overhead. Bumping the container size without thinking about the split usually just grows framework overhead. Read the Flink [memory tuning guide](https://nightlies.apache.org/flink/flink-docs-master/docs/deployment/memory/mem_setup/) before scaling up. +- **CPU per TaskManager × slots per TaskManager = parallelism budget.** Setting `taskmanager.numberOfTaskSlots: 4` on a 4 CPU pod means 4 parallel tasks per TM. Balance is workload-dependent: shuffle-heavy jobs like 1 slot per pod (fewer noisy-neighbour effects); CPU-light jobs pack more slots. +- **RocksDB needs disk, not just memory.** When you swap to RocksDB, attach a PVC on `flink-gp3` and set `state.backend.rocksdb.localdir` to the mount path. Local state grows with keyspace × timers. +- **Checkpoint interval is a knob, not a constant.** Default is a few seconds. Shorter = smaller recovery gap but higher I/O overhead. Longer = less overhead but more work to redo on restart. +- **Instance families:** the shipped `m5a`/`r5a` are the workshop starting point. `r7iz` (Sapphire Rapids, high sustained clock) for latency-critical stream processing. `m7g` (Graviton3) for the best price-performance if all your operators and connectors are ARM-compatible. + +## Extending + +- **Kafka source.** The obvious follow-up: swap `StateMachineExample.jar` for a Flink job that reads from `cluster-kafka-bootstrap.kafka.svc:9092`. The Flink Kafka connector is a bundled dependency in the `flink:1.20` image. +- **ClickHouse sink.** Once your Flink job produces aggregated records, sink them into ClickHouse via the [flink-connector-jdbc](https://nightlies.apache.org/flink/flink-docs-master/docs/connectors/datastream/jdbc/) using the aggregated HTTP endpoint at `http://clickhouse-cluster.clickhouse.svc:8123`. That's the canonical Kafka → Flink → ClickHouse streaming pipeline. +- **HA JobManager.** Set `spec.jobManager.replicas: 2` and add a `kubernetesHAOptions` block so JobManager metadata is persisted in a ConfigMap and TaskManagers can reconnect if the leader dies. Requires a persistent checkpoint store (S3, not local). +- **Autoscaling.** The Flink operator 1.15+ supports the [Autoscaler](https://nightlies.apache.org/flink/flink-kubernetes-operator-docs-main/docs/custom-resource/autoscaler/) — scale TaskManagers based on backpressure and lag rather than by hand. Enable via `spec.flinkConfiguration.job.autoscaler.enabled: "true"`. +- **Blue/Green deployments.** Flink operator 1.15's headline feature: deploy a new version of your streaming app in parallel with the running one, switch over on savepoint. See the [Blue/Green docs](https://nightlies.apache.org/flink/flink-kubernetes-operator-docs-main/docs/custom-resource/blue-green/) for the operator-native pattern. +- **Disable the operator install.** Set `enable_flink_lab = false` in Terraform. The Flink operator and `flink-gp3` StorageClass are no longer created; the Flink Karpenter NodePool remains (harmless — nothing tolerates its taint). diff --git a/analytics/flink/cleanup.sh b/analytics/flink/cleanup.sh new file mode 100755 index 000000000..7cd66286f --- /dev/null +++ b/analytics/flink/cleanup.sh @@ -0,0 +1,32 @@ +#!/usr/bin/env bash +set -euo pipefail + +# Tears down the sample FlinkDeployment. +# The Flink Kubernetes Operator, the flink-gp3 StorageClass, the +# dedicated Flink NodePool, and the flink namespace are all managed +# by Terraform (enable_flink_lab = true) and are removed by +# `terraform destroy`. This script does NOT touch them. + +NAMESPACE=flink + +echo "Deleting FlinkDeployment (drains job gracefully; savepoint if any)..." +kubectl delete flinkdeployment state-machine -n "${NAMESPACE}" --ignore-not-found + +echo "" +echo "Waiting up to 3 minutes for JobManager + TaskManager pods to terminate..." +kubectl -n "${NAMESPACE}" wait --for=delete pod \ + -l app=state-machine --timeout=180s || true + +echo "" +echo "Deleting any persistent volume claims (only present if the job used PVCs for RocksDB state)..." +# Retain-policy PVs survive PVC deletion; the underlying EBS volumes have +# to be released separately if you want to stop paying for them. The +# workshop's `terraform destroy` handles that when it removes the +# StorageClass and namespace at the end. +kubectl -n "${NAMESPACE}" delete pvc --all --ignore-not-found + +echo "" +echo "Flink job resources removed. The Flink Kubernetes Operator," +echo "flink-gp3 StorageClass, and Karpenter NodePool remain" +echo "(Terraform-managed). Run 'terraform destroy' in" +echo "analytics/terraform/spark-k8s-operator/ to tear down the whole workshop." diff --git a/analytics/flink/deploy-flink.sh b/analytics/flink/deploy-flink.sh new file mode 100755 index 000000000..7f258636a --- /dev/null +++ b/analytics/flink/deploy-flink.sh @@ -0,0 +1,54 @@ +#!/usr/bin/env bash +set -euo pipefail + +# Deploys the sample FlinkDeployment onto the workshop's EKS cluster. +# Prerequisites are managed by Terraform when enable_flink_lab = true: +# - Apache Flink Kubernetes Operator running in the "flink" namespace +# - "flink-gp3" StorageClass +# - Dedicated Flink Karpenter NodePool with +# workload=flink:NoSchedule taint + +SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" +NAMESPACE=flink + +echo "Preflight: verifying operator, storageclass, and NodePool are in place..." +if ! kubectl -n "${NAMESPACE}" rollout status deploy/flink-kubernetes-operator --timeout=120s; then + echo "ERROR: Flink Kubernetes Operator not ready in '${NAMESPACE}'." + echo " Re-apply Terraform with enable_flink_lab=true — the operator," + echo " StorageClass, and NodePool all come from that single toggle." + exit 1 +fi + +if ! kubectl get storageclass flink-gp3 &>/dev/null; then + echo "ERROR: StorageClass 'flink-gp3' not found." + echo " Re-apply Terraform with enable_flink_lab=true." + exit 1 +fi + +if ! kubectl get nodepool.karpenter.sh flink &>/dev/null; then + echo "ERROR: Karpenter NodePool 'flink' not found." + echo " Re-apply Terraform (manifests/automode/nodepool-flink.yaml)." + exit 1 +fi + +echo "" +echo "Applying FlinkDeployment (StateMachineExample, parallelism 2)..." +kubectl apply -f "${SCRIPT_DIR}/flink-cluster.yaml" + +echo "" +echo "Waiting for FlinkDeployment to reach STABLE (3-6 minutes typical while nodes provision)..." +# jsonpath wait against LIFECYCLE STATE. STABLE means the operator has +# reconciled the CR into a healthy JM + N TMs with the job RUNNING. +kubectl -n "${NAMESPACE}" wait --for=jsonpath='{.status.lifecycleState}'=STABLE \ + flinkdeployment/state-machine --timeout=600s + +echo "" +kubectl get flinkdeployment -n "${NAMESPACE}" +echo "" +kubectl get pods -n "${NAMESPACE}" -o wide +echo "" +echo "JobManager REST endpoint (in-cluster):" +echo " http://state-machine-rest.${NAMESPACE}.svc.cluster.local:8081" +echo "" +echo "Port-forward the Flink Web UI:" +echo " kubectl -n ${NAMESPACE} port-forward svc/state-machine-rest 8081:8081" diff --git a/analytics/flink/flink-cluster.yaml b/analytics/flink/flink-cluster.yaml new file mode 100644 index 000000000..eeba18de6 --- /dev/null +++ b/analytics/flink/flink-cluster.yaml @@ -0,0 +1,96 @@ +--- +# --------------------------------------------------------------------- +# Sample FlinkDeployment — StateMachine example +# --------------------------------------------------------------------- +# The StateMachine example ships in the official flink:1.20 image at +# `local:///opt/flink/examples/streaming/StateMachineExample.jar`. +# It generates its own stream of events, applies a stateful KeyedProcess +# operator, and writes results to stdout — zero external dependencies, +# ideal for demonstrating the operator without wiring up Kafka + a sink. +# +# Deployed in "Application mode": one JobManager plus N TaskManagers +# per FlinkDeployment. This is the modern operator-native pattern +# (Session mode — one Flink cluster hosting many jobs — is legacy). +# +# Pod placement: +# - nodeSelector `workload=flink` pins pods to the dedicated NodePool. +# - Toleration for `workload=flink:NoSchedule` gets past the taint. +# - No AZ anti-affinity is set: this is a single-job demo, not an HA +# multi-replica setup. For HA JobManager, add +# `spec.jobManager.replicas: 2` plus a `kubernetesHAOptions` block +# — see the workshop README for the upgrade path. +# --------------------------------------------------------------------- +apiVersion: flink.apache.org/v1beta1 +kind: FlinkDeployment +metadata: + name: state-machine + namespace: flink +spec: + image: flink:1.20 + flinkVersion: v1_20 + + flinkConfiguration: + # 1 slot per TaskManager, 2 TaskManagers → total parallelism 2 + # (matches spec.job.parallelism below). + taskmanager.numberOfTaskSlots: "1" + + # HashMap state backend is fine for the demo. Swap to + # state.backend.type: rocksdb + # once state outgrows the JVM heap; then also attach a PVC to the + # TaskManager pod on the flink-gp3 StorageClass for the local RocksDB dir. + state.backend.type: hashmap + state.checkpoints.dir: file:///tmp/flink-checkpoints + state.savepoints.dir: file:///tmp/flink-savepoints + + # Service account bundled with the operator install; carries the RBAC + # the JobManager needs to spawn TaskManager pods via Flink's native + # Kubernetes integration. + serviceAccount: flink + + jobManager: + resource: + memory: 2048m + cpu: 1 + podTemplate: + apiVersion: v1 + kind: Pod + metadata: + name: jobmanager-pod-template + spec: + nodeSelector: + workload: flink + tolerations: + - key: workload + operator: Equal + value: flink + effect: NoSchedule + containers: + - name: flink-main-container + + taskManager: + replicas: 2 + resource: + memory: 2048m + cpu: 1 + podTemplate: + apiVersion: v1 + kind: Pod + metadata: + name: taskmanager-pod-template + spec: + nodeSelector: + workload: flink + tolerations: + - key: workload + operator: Equal + value: flink + effect: NoSchedule + containers: + - name: flink-main-container + + job: + # The StateMachine example is baked into the official flink:1.20 image. + jarURI: local:///opt/flink/examples/streaming/StateMachineExample.jar + parallelism: 2 + upgradeMode: stateless + state: running diff --git a/analytics/terraform/spark-k8s-operator/flink-operator.tf b/analytics/terraform/spark-k8s-operator/flink-operator.tf new file mode 100644 index 000000000..b5a22bd5c --- /dev/null +++ b/analytics/terraform/spark-k8s-operator/flink-operator.tf @@ -0,0 +1,113 @@ +#--------------------------------------------------------------- +# Flink lab — Apache Flink Kubernetes Operator + tuned StorageClass +#--------------------------------------------------------------- +# This file installs the two pieces of shared infrastructure that the +# Flink lab depends on. The FlinkDeployment itself is a custom resource +# that the participant applies during the lab (see `analytics/flink/`), +# so JobManager / TaskManager capacity is only provisioned when someone +# actually runs the lab. +# +# 1. Apache Flink Kubernetes Operator (Helm) — watches for +# `FlinkDeployment` and `FlinkSessionJob` CRs and reconciles them +# into JobManager Deployments plus TaskManager pods managed via +# Flink's native Kubernetes integration. +# +# 2. `flink-gp3` StorageClass — tuned gp3 for Flink stateful state +# backends (RocksDB checkpoints, incremental savepoints). 6000 IOPS +# / 500 MiB/s per volume; `reclaimPolicy: Retain` because savepoint +# state is precious. +# +# Cert-manager is required for the Flink operator's admission webhooks. +# The workshop cluster already ships cert-manager as part of its addon +# stack, so this file does not install it. +# +# The dedicated Flink Karpenter NodePool lives in +# `manifests/automode/nodepool-flink.yaml`; it is picked up automatically +# by the `auto_mode_nodepools` fileset() discovery in `eks.tf`, so it +# deploys regardless of `enable_flink_lab` (an untainted NodePool with +# no matching pods is inert and costs nothing). +# +# Toggle the operator + StorageClass with `var.enable_flink_lab` +# (default true). +#--------------------------------------------------------------- + +locals { + flink_lab = { + namespace = "flink" + operator_version = var.flink_operator_version + storage_class_name = "flink-gp3" + # Flink's checkpoint uploads are bursty (each interval, every + # TaskManager writes its state delta). 6000 IOPS + 500 MiB/s covers + # the common workshop shape; graduate to `io2` or bigger `gp3` if + # your job's per-checkpoint state runs into the GB range. + storage_class_iops = 6000 + storage_class_throughput_mib = 500 + } +} + +resource "helm_release" "flink_kubernetes_operator" { + count = var.enable_flink_lab ? 1 : 0 + + name = "flink-kubernetes-operator" + namespace = local.flink_lab.namespace + create_namespace = true + # Apache publishes a Helm chart per operator version at + # https://downloads.apache.org/flink/flink-kubernetes-operator-/ + # The chart name is always `flink-kubernetes-operator`. + repository = "https://downloads.apache.org/flink/flink-kubernetes-operator-${local.flink_lab.operator_version}/" + chart = "flink-kubernetes-operator" + version = local.flink_lab.operator_version + timeout = 600 + + # The operator watches all namespaces by default and installs + # cluster-scoped CRDs. Participants apply FlinkDeployment CRs into + # the `flink` namespace in the lab, but the operator's RBAC allows + # experimentation elsewhere. + depends_on = [ + module.eks, + kubectl_manifest.auto_mode_nodepools, + ] +} + +# Tuned gp3 StorageClass for Flink state backend volumes. +# +# Why the parameters: +# - `iops: 6000` — RocksDB compaction and checkpoint uploads are +# write-heavy in bursts. gp3 default 3000 IOPS gets choked under +# load; 6000 keeps us clear of the Kafka + ClickHouse allocation +# ceilings while giving enough headroom for typical workshop jobs. +# - `throughput: 500` MiB/s — enough for savepoint uploads to +# s3-compatible checkpoint stores without saturating the per-volume +# EBS baseline. +# - `reclaimPolicy: Retain` — savepoint volumes are precious. A +# stray `kubectl delete pvc` on `Delete` policy would silently take +# the volume with it. Retain requires an explicit release step. +# - `allowVolumeExpansion: true` — in-place PVC resize when state +# grows past the initial size, no rebuild required. +# - `WaitForFirstConsumer` — EBS volumes are AZ-scoped; deferring +# provisioning until the pod is scheduled avoids AZ mismatches. +resource "kubectl_manifest" "flink_gp3_storageclass" { + count = var.enable_flink_lab ? 1 : 0 + + yaml_body = yamlencode({ + apiVersion = "storage.k8s.io/v1" + kind = "StorageClass" + metadata = { + name = local.flink_lab.storage_class_name + } + provisioner = "ebs.csi.eks.amazonaws.com" + volumeBindingMode = "WaitForFirstConsumer" + reclaimPolicy = "Retain" + allowVolumeExpansion = true + parameters = { + type = "gp3" + fsType = "xfs" + encrypted = "true" + iops = tostring(local.flink_lab.storage_class_iops) + throughput = tostring(local.flink_lab.storage_class_throughput_mib) + } + }) + wait = true + + depends_on = [module.eks] +} diff --git a/analytics/terraform/spark-k8s-operator/manifests/automode/nodepool-flink.yaml b/analytics/terraform/spark-k8s-operator/manifests/automode/nodepool-flink.yaml new file mode 100644 index 000000000..d9cc8b267 --- /dev/null +++ b/analytics/terraform/spark-k8s-operator/manifests/automode/nodepool-flink.yaml @@ -0,0 +1,80 @@ +# --------------------------------------------------------------------- +# Dedicated Flink NodePool +# --------------------------------------------------------------------- +# Flink jobs are stateful long-running streaming processes. A Flink +# JobManager loss triggers a job restart (with recovery from the latest +# checkpoint); a TaskManager loss triggers partition reassignment and +# a rescale. Both hurt latency and cost work. +# +# Two-way isolation with the Flink pods: +# - Node label `workload=flink` — used for observability/selectors. +# - Taint `workload=flink:NoSchedule` — nothing else in the cluster +# tolerates this, so nothing else lands here. +# - The FlinkDeployment CR (analytics/flink/flink-cluster.yaml) +# carries the matching toleration and nodeSelector on both the +# JobManager and TaskManager pod templates. +# +# Consolidation: +# - WhenEmpty with a 5-minute delay. Voluntary consolidation of a +# partially-loaded Flink node causes a job restart. Karpenter still +# handles expiry / drift / forced disruption. +# +# Instance selection: +# - m/r families, Gen 5+, Nitro. Flink is memory-heavy (JVM heap + +# off-heap for RocksDB when enabled) and CPU-heavy during joins +# and windowed aggregations. r-family is the natural default; +# m stays in the mix for smaller footprints. +# - amd64 only. The Flink image and its extension JARs are amd64- +# first — arm64 support exists but keeps things predictable across +# the whole stack. +# - Sizes 2xlarge..8xlarge — a typical workshop-shape Flink job +# fits in 2xlarge; larger sizes let a participant crank parallelism. +# - T3 excluded — CPU credits deplete under sustained streaming load. +# +# For heavier workloads: `r7iz` (Sapphire Rapids high clock speed) for +# latency-critical stream processing, or `m7g` (Graviton3) for the best +# price-performance if all your operators are ARM-compatible. +# --------------------------------------------------------------------- +apiVersion: karpenter.sh/v1 +kind: NodePool +metadata: + name: flink +spec: + template: + metadata: + labels: + workload: flink + spec: + taints: + - key: workload + value: flink + effect: NoSchedule + nodeClassRef: + group: eks.amazonaws.com + kind: NodeClass + name: ebs-gp3-1000gi-6000iops-1000tp + requirements: + - key: karpenter.sh/capacity-type + operator: In + values: ["on-demand"] + - key: kubernetes.io/arch + operator: In + values: ["amd64"] + - key: eks.amazonaws.com/instance-category + operator: In + values: ["m", "r"] + - key: eks.amazonaws.com/instance-size + operator: In + values: ["2xlarge", "4xlarge", "8xlarge"] + - key: eks.amazonaws.com/instance-hypervisor + operator: In + values: ["nitro"] + - key: eks.amazonaws.com/instance-generation + operator: Gt + values: ["4"] + limits: + cpu: "200" + memory: 800Gi + disruption: + consolidationPolicy: WhenEmpty + consolidateAfter: 5m diff --git a/analytics/terraform/spark-k8s-operator/variables.tf b/analytics/terraform/spark-k8s-operator/variables.tf index 23cbe3535..4ec2a43e5 100644 --- a/analytics/terraform/spark-k8s-operator/variables.tf +++ b/analytics/terraform/spark-k8s-operator/variables.tf @@ -109,3 +109,15 @@ variable "enable_celeborn" { type = bool default = true } + +variable "enable_flink_lab" { + description = "Enable the Apache Flink lab. When true, Terraform installs the Apache Flink Kubernetes Operator into the 'flink' namespace and creates the 'flink-gp3' StorageClass. The dedicated Flink Karpenter NodePool ('nodepool-flink.yaml') is applied regardless — it is inert without Flink pods." + type = bool + default = true +} + +variable "flink_operator_version" { + description = "Apache Flink Kubernetes Operator Helm chart version. The Helm repository URL is derived from this value so both stay in lockstep. 1.15.0 is the current stable release published to https://downloads.apache.org/flink/." + type = string + default = "1.15.0" +}