diff --git a/.github/workflows/ci-jdk11.yml b/.github/workflows/ci-jdk11.yml
index e5878aaa2..26e0991b6 100644
--- a/.github/workflows/ci-jdk11.yml
+++ b/.github/workflows/ci-jdk11.yml
@@ -69,6 +69,12 @@ jobs:
java-version: '11'
distribution: 'temurin'
+ - name: Verify Docker environment
+ timeout-minutes: 2
+ run: |
+ python3 tools/ci/test_check_docker.py
+ bash tools/ci/check-docker.sh
+
- name: Setup Protoc
uses: arduino/setup-protoc@v2
with:
diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index bba2534f6..217afc9c7 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -72,6 +72,12 @@ jobs:
distribution: 'temurin'
cache: maven
+ - name: Verify Docker environment
+ timeout-minutes: 2
+ run: |
+ python3 tools/ci/test_check_docker.py
+ bash tools/ci/check-docker.sh
+
- name: Setup Protoc
uses: arduino/setup-protoc@v2
with:
diff --git a/geaflow-ai/src/test/java/org/apache/geaflow/ai/index/KeywordIndexLifecycleGoldenTest.java b/geaflow-ai/src/test/java/org/apache/geaflow/ai/index/KeywordIndexLifecycleGoldenTest.java
index 255681ad5..d86e9b8bd 100644
--- a/geaflow-ai/src/test/java/org/apache/geaflow/ai/index/KeywordIndexLifecycleGoldenTest.java
+++ b/geaflow-ai/src/test/java/org/apache/geaflow/ai/index/KeywordIndexLifecycleGoldenTest.java
@@ -39,23 +39,24 @@
import org.apache.geaflow.ai.graph.io.VertexSchema;
import org.apache.geaflow.ai.index.vector.KeywordVector;
import org.apache.geaflow.ai.operator.GraphSearchStore;
+import org.apache.lucene.queryparser.classic.ParseException;
+import org.apache.lucene.store.AlreadyClosedException;
+import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
/**
- * Lifecycle and retrieval-semantics tests for the keyword index, pinning the behavior issue #856
- * calls out as fragile: the store is rebuilt during every query, retrieval tolerates common-token
- * noise, and deleted entities are filtered at query time rather than removed from the index.
+ * Lifecycle and retrieval-semantics tests for the keyword index: committed writes remain
+ * searchable until the store is closed, retrieval tolerates common-token noise, and deleted
+ * entities are filtered at query time rather than removed from the index.
*
*
The usage contract these tests document, in one place:
*
*
- *
Build, close, search, discard. A query session indexes what it wants searchable,
- * closes the store (which commits the segments) and only then searches. Searching before
- * close silently returns nothing — the {@code IndexNotFoundException} of the uncommitted
- * directory is swallowed. Indexing again after close fails outright, because
- * {@code SearchStore} reuses one {@code IndexWriterConfig} across writers. The store is
- * therefore single-shot, and {@code SessionOperator} exercises exactly this order.
+ *
Build, finish writing, search, close. A query session indexes its entities and calls
+ * {@link GraphSearchStore#finishWriting()} to commit the segments while retaining the
+ * directory and analyzer for reads. Searching before the first commit returns no results.
+ * Closing the store releases its resources and rejects subsequent reads and writes.
*
Stop words are searchable. The analyzer is built with an empty stop-word set
* (see {@link KeywordTokenizationGoldenTest}), so {@code the} matches documents
* containing {@code the}.
@@ -76,17 +77,26 @@ public class KeywordIndexLifecycleGoldenTest {
private static final String LABEL = "chunk";
private static final String EDGE_LABEL = "rel";
- /** The production order: index, close, search. Everything visible before close is found. */
+ private final List stores = new ArrayList<>();
+
+ @AfterEach
+ public void closeStores() {
+ for (GraphSearchStore store : stores) {
+ store.close();
+ }
+ }
+
+ /** Finished writes remain searchable until the store is closed. */
@Test
- public void indexedEntitiesAreSearchableAfterClose() {
- LocalMemoryGraphAccessor accessor = graph("v1");
- GraphSearchStore store = new GraphSearchStore();
+ public void indexedEntitiesAreSearchableAfterFinishWriting() {
+ final LocalMemoryGraphAccessor accessor = graph("v1");
+ GraphSearchStore store = createStore();
indexVertex(store, "v1", "alpaca", "grazes");
- store.close();
+ store.finishWriting();
List found = searchIds(store, accessor, "alpaca");
Assertions.assertEquals(Collections.singletonList("v1"), found,
- "a term indexed before close must be found after close");
+ "a term indexed before finishWriting must be found after finishWriting");
Assertions.assertEquals(Collections.singletonList("v1"), searchIds(store, accessor, "ALPACA"),
"matching is case-insensitive on both sides of the index");
Assertions.assertEquals(Collections.singletonList("v1"), searchIds(store, accessor, "grazes"),
@@ -96,49 +106,58 @@ public void indexedEntitiesAreSearchableAfterClose() {
}
/**
- * Searching before close returns an empty list instead of failing. This is why the
- * close-before-search order above is a contract rather than a convention: getting it wrong
- * fails silently.
+ * The first commit makes buffered documents visible; an earlier empty search must not
+ * prevent the same store from finding those documents after finishWriting.
*/
@Test
- public void searchBeforeCloseReturnsEmptyResults() {
- LocalMemoryGraphAccessor accessor = graph("v1");
- GraphSearchStore store = new GraphSearchStore();
+ public void searchBeforeFinishWritingReturnsEmptyResults() {
+ final LocalMemoryGraphAccessor accessor = graph("v1");
+ GraphSearchStore store = createStore();
indexVertex(store, "v1", "alpaca");
Assertions.assertEquals(Collections.emptyList(), searchIds(store, accessor, "alpaca"),
- "search before close must return empty, not the committed documents (there are none)");
+ "an uncommitted index has no searchable documents");
+ store.finishWriting();
+ Assertions.assertEquals(Collections.singletonList("v1"), searchIds(store, accessor, "alpaca"));
}
- /**
- * Indexing after close throws: {@code SearchStore} hands the same {@code IndexWriterConfig}
- * to a second {@code IndexWriter} and Lucene rejects that. Combined with the test above this
- * fixes the lifecycle as index-then-close-then-search with no reopen — the resident index of
- * PR #825 is the proposed way out of this single-shot shape.
- */
+ /** Closing is terminal and rejects subsequent indexing at the store boundary. */
@Test
public void indexingAfterCloseFails() {
- LocalMemoryGraphAccessor accessor = graph("v1");
- GraphSearchStore store = new GraphSearchStore();
+ GraphSearchStore store = createStore();
indexVertex(store, "v1", "alpaca");
store.close();
- RuntimeException failure = Assertions.assertThrows(RuntimeException.class,
- () -> indexVertex(store, "v2", "lama"),
- "indexing into a closed store must fail loudly rather than lose the document");
- Assertions.assertTrue(failure.getCause() instanceof IllegalStateException,
- "the underlying refusal is Lucene's IndexWriterConfig sharing guard, got: "
- + failure.getCause());
+ IllegalStateException failure = Assertions.assertThrows(IllegalStateException.class,
+ () -> indexVertex(store, "v2", "lama"));
+ Assertions.assertEquals("graph search store is closed", failure.getMessage());
+ }
+
+ @Test
+ public void closedStoreRejectsReadsAndFinishingAndReleasesDirectory() {
+ final LocalMemoryGraphAccessor accessor = graph("v1");
+ GraphSearchStore store = createStore();
+ indexVertex(store, "v1", "alpaca");
+ store.finishWriting();
+ store.finishWriting();
+ Assertions.assertEquals(Collections.singletonList("v1"), searchIds(store, accessor, "alpaca"));
+ store.close();
+ store.close();
+ Assertions.assertThrows(IllegalStateException.class, () -> store.search("alpaca", accessor));
+ Assertions.assertThrows(IllegalStateException.class, store::finishWriting);
+ Assertions.assertThrows(AlreadyClosedException.class, () -> store.getDirectory().listAll());
}
/** An empty query reaches the query parser and surfaces as the store's read failure. */
@Test
public void emptyQueryFailsTheSearchStore() {
- LocalMemoryGraphAccessor accessor = graph("v1");
- GraphSearchStore store = new GraphSearchStore();
+ final LocalMemoryGraphAccessor accessor = graph("v1");
+ GraphSearchStore store = createStore();
indexVertex(store, "v1", "alpaca");
- store.close();
+ store.finishWriting();
RuntimeException failure = Assertions.assertThrows(RuntimeException.class,
() -> store.search("", accessor));
Assertions.assertEquals("Cannot read search store", failure.getMessage());
+ Assertions.assertTrue(failure.getCause() instanceof ParseException,
+ "an empty query must reach the parser while the store is still open");
}
/**
@@ -148,11 +167,11 @@ public void emptyQueryFailsTheSearchStore() {
*/
@Test
public void multiWordQueriesMatchAnyTerm() {
- LocalMemoryGraphAccessor accessor = graph("v1", "v2");
- GraphSearchStore store = new GraphSearchStore();
+ final LocalMemoryGraphAccessor accessor = graph("v1", "v2");
+ GraphSearchStore store = createStore();
indexVertex(store, "v1", "alpaca");
indexVertex(store, "v2", "grazes");
- store.close();
+ store.finishWriting();
Assertions.assertEquals(Collections.singletonList("v1"), searchIds(store, accessor, "alpaca.lama"),
"OR semantics: only one of the two terms needs to be present");
@@ -161,10 +180,10 @@ public void multiWordQueriesMatchAnyTerm() {
/** Stop words are ordinary searchable terms (the analyzer has no stop-word list). */
@Test
public void stopWordsAreSearchableTerms() {
- LocalMemoryGraphAccessor accessor = graph("v1");
- GraphSearchStore store = new GraphSearchStore();
+ final LocalMemoryGraphAccessor accessor = graph("v1");
+ GraphSearchStore store = createStore();
indexVertex(store, "v1", "the", "fox");
- store.close();
+ store.finishWriting();
Assertions.assertEquals(Collections.singletonList("v1"), searchIds(store, accessor, "the"),
"with an empty stop-word set, 'the' indexes and matches like any other word");
}
@@ -179,11 +198,11 @@ public void stopWordsAreSearchableTerms() {
*/
@Test
public void unicodeKeywordsMatchByCharacter() {
- LocalMemoryGraphAccessor accessor = graph("v1", "v2");
- GraphSearchStore store = new GraphSearchStore();
+ final LocalMemoryGraphAccessor accessor = graph("v1", "v2");
+ GraphSearchStore store = createStore();
indexVertex(store, "v1", "学习", "效率");
indexVertex(store, "v2", "习题", "效率");
- store.close();
+ store.finishWriting();
Assertions.assertEquals(Arrays.asList("v1", "v2"), searchIds(store, accessor, "效率"),
"an exact two-character keyword matches the documents carrying it");
@@ -211,15 +230,15 @@ public void commonWrapperTokensDoNotCrowdOutExactAnchoring() {
for (int i = 0; i < corpusSize; i++) {
ids[i] = "v" + i;
}
- LocalMemoryGraphAccessor accessor = graph(ids);
+ final LocalMemoryGraphAccessor accessor = graph(ids);
- GraphSearchStore store = new GraphSearchStore();
+ GraphSearchStore store = createStore();
for (int i = 0; i < corpusSize; i++) {
// One shared filler keyword plus a unique one: every document overlaps its neighbors,
// none of them is distinguishable through the filler alone.
indexVertex(store, ids[i], "shared", "unique" + i);
}
- store.close();
+ store.finishWriting();
Assertions.assertEquals(Collections.singletonList("v17"),
searchIds(store, accessor, "unique17"),
@@ -234,21 +253,20 @@ public void commonWrapperTokensDoNotCrowdOutExactAnchoring() {
}
/**
- * Incremental addition is accumulation within one open session: documents added in several
- * batches are all committed by the single close and all searchable. There is no incremental
- * visibility into an already-searched index (see {@link #indexingAfterCloseFails()}).
+ * Documents added in several batches within one write session are committed together by
+ * finishWriting and remain searchable until close.
*/
@Test
- public void incrementalAddsAccumulateUntilClose() {
- LocalMemoryGraphAccessor accessor = graph("v1", "v2", "v3");
- GraphSearchStore store = new GraphSearchStore();
+ public void incrementalAddsAccumulateUntilFinishWriting() {
+ final LocalMemoryGraphAccessor accessor = graph("v1", "v2", "v3");
+ GraphSearchStore store = createStore();
indexVertex(store, "v1", "alpaca");
indexVertex(store, "v2", "lama");
indexVertex(store, "v3", "zebra");
- store.close();
+ store.finishWriting();
Assertions.assertEquals(Arrays.asList("v1", "v2", "v3"), searchIds(store, accessor, "alpaca lama zebra"),
- "every batch added before close is committed and searchable");
+ "every batch added before finishWriting is committed and searchable");
}
/**
@@ -258,8 +276,8 @@ public void incrementalAddsAccumulateUntilClose() {
*/
@Test
public void deletedEntitiesAreFilteredAtQueryTime() {
- LocalMemoryGraphAccessor accessor = graph("v1");
- GraphSearchStore store = new GraphSearchStore();
+ final LocalMemoryGraphAccessor accessor = graph("v1");
+ GraphSearchStore store = createStore();
indexVertex(store, "v1", "alpaca");
// A vertex indexed with a registered label but an id the graph does not hold — the shape
// a deleted entity leaves behind.
@@ -268,7 +286,7 @@ public void deletedEntitiesAreFilteredAtQueryTime() {
store.indexVertex(new GraphVertex(
new Vertex("unknown", "v9", Collections.singletonList("x"))),
Collections.singletonList(new KeywordVector("orphanword")));
- store.close();
+ store.finishWriting();
Assertions.assertEquals(Collections.singletonList("v1"), searchIds(store, accessor, "alpaca"));
Assertions.assertTrue(searchIds(store, accessor, "ghostword").isEmpty(),
@@ -279,6 +297,12 @@ public void deletedEntitiesAreFilteredAtQueryTime() {
// ------------------------------------------------------------------ helpers
+ private GraphSearchStore createStore() {
+ GraphSearchStore store = new GraphSearchStore();
+ stores.add(store);
+ return store;
+ }
+
private LocalMemoryGraphAccessor graph(String... ids) {
GraphSchema schema = new GraphSchema();
schema.setName("keyword-golden");
diff --git a/geaflow-ai/src/test/java/org/apache/geaflow/ai/index/KeywordIndexSerializationGoldenTest.java b/geaflow-ai/src/test/java/org/apache/geaflow/ai/index/KeywordIndexSerializationGoldenTest.java
index 91c4f275c..f5a4ab82f 100644
--- a/geaflow-ai/src/test/java/org/apache/geaflow/ai/index/KeywordIndexSerializationGoldenTest.java
+++ b/geaflow-ai/src/test/java/org/apache/geaflow/ai/index/KeywordIndexSerializationGoldenTest.java
@@ -59,42 +59,46 @@ public class KeywordIndexSerializationGoldenTest {
private static final String SERIALIZATION_GOLDEN = "/index/keyword/serialized_docs_golden.txt";
/**
- * Indexes one vertex, one edge and one vertex carrying two vectors, closes the store the way
- * the production path does, then reads every document back field by field and compares against
+ * Indexes one vertex, one edge and one vertex carrying two vectors, finishes writing, then
+ * reads every document back before releasing the store and compares against
* the golden file. Rendering is {@code field=[value] ...} with the production field
* order (vertices id/label/content, edges src/dst/label/content).
*/
@Test
public void indexedDocumentsMatchGolden() throws IOException {
GraphSearchStore store = new GraphSearchStore();
- store.indexVertex(new GraphVertex(
- new Vertex("chunk", "v1", Collections.singletonList("the alpaca grazes"))),
- Collections.singletonList(new KeywordVector("alpaca", "grazes")));
- store.indexEdge(new GraphEdge(
- new Edge("rel", "v1", "v2", Collections.singletonList("x"))),
- Collections.singletonList(new KeywordVector("master")));
- store.indexVertex(new GraphVertex(
- new Vertex("chunk", "v3", Collections.singletonList("two vectors"))),
- Arrays.asList(new KeywordVector("a"), new KeywordVector("b")));
- store.close();
+ try {
+ store.indexVertex(new GraphVertex(
+ new Vertex("chunk", "v1", Collections.singletonList("the alpaca grazes"))),
+ Collections.singletonList(new KeywordVector("alpaca", "grazes")));
+ store.indexEdge(new GraphEdge(
+ new Edge("rel", "v1", "v2", Collections.singletonList("x"))),
+ Collections.singletonList(new KeywordVector("master")));
+ store.indexVertex(new GraphVertex(
+ new Vertex("chunk", "v3", Collections.singletonList("two vectors"))),
+ Arrays.asList(new KeywordVector("a"), new KeywordVector("b")));
+ store.finishWriting();
- List actual = new ArrayList<>();
- try (DirectoryReader reader = DirectoryReader.open(store.getDirectory())) {
- for (int i = 0; i < reader.maxDoc(); i++) {
- Document doc = reader.document(i);
- if (doc.get("id") != null) {
- actual.add(String.format("vertex id=[%s] label=[%s] content=[%s]",
- doc.get("id"), doc.get("label"), doc.get("content")));
- } else {
- actual.add(String.format("edge src=[%s] dst=[%s] label=[%s] content=[%s]",
- doc.get("src"), doc.get("dst"), doc.get("label"), doc.get("content")));
+ List actual = new ArrayList<>();
+ try (DirectoryReader reader = DirectoryReader.open(store.getDirectory())) {
+ for (int i = 0; i < reader.maxDoc(); i++) {
+ Document doc = reader.document(i);
+ if (doc.get("id") != null) {
+ actual.add(String.format("vertex id=[%s] label=[%s] content=[%s]",
+ doc.get("id"), doc.get("label"), doc.get("content")));
+ } else {
+ actual.add(String.format("edge src=[%s] dst=[%s] label=[%s] content=[%s]",
+ doc.get("src"), doc.get("dst"), doc.get("label"), doc.get("content")));
+ }
}
}
- }
- Assertions.assertEquals(goldenLines(), actual,
- "the serialized form of indexed keyword vectors changed; review against "
- + SERIALIZATION_GOLDEN);
+ Assertions.assertEquals(goldenLines(), actual,
+ "the serialized form of indexed keyword vectors changed; review against "
+ + SERIALIZATION_GOLDEN);
+ } finally {
+ store.close();
+ }
}
private static List goldenLines() throws IOException {
diff --git a/geaflow-kubernetes-operator/Dockerfile b/geaflow-kubernetes-operator/Dockerfile
index 6c4539808..0376b793e 100644
--- a/geaflow-kubernetes-operator/Dockerfile
+++ b/geaflow-kubernetes-operator/Dockerfile
@@ -17,34 +17,15 @@
# under the License.
#
-# base image
-FROM docker.io/library/centos:7@sha256:9d4bcbbb213dfd745b58be38b13b996ebb5ac315fe75711bd618426a630e0987
+# Java 11 runtime; the operator does not execute job JVMs in its own container.
+FROM docker.io/library/eclipse-temurin:11-jre-jammy@sha256:0c567a8bd671331808b89a96976fabb305d57b123dec4fa62a20bfb89f2b32b9
VOLUME /tmp
ENV GEAFLOW_INSTALL_PATH=/opt
ENV GEAFLOW_HOME=$GEAFLOW_INSTALL_PATH/geaflow
-# base packages
-RUN yum -y install wget ps vim curl
-
-# jdk11
-RUN yum -y install java-11-openjdk java-11-openjdk-devel \
- && echo "export JAVA_HOME=/usr/lib/jvm/java-11-openjdk" >> /etc/profile \
- && echo "export JRE_HOME=\${JAVA_HOME}/jre" >> /etc/profile \
- && echo "export CLASSPATH=.:\${JAVA_HOME}/lib:\${JRE_HOME}/lib:\$CLASSPATH" >> /etc/profile \
- && echo "export JAVA_PATH=\${JAVA_HOME}/bin:\${JRE_HOME}/bin" >> /etc/profile \
- && echo "export PATH=\$PATH:\${JAVA_PATH}" >> /etc/profile \
- && source /etc/profile
-
-# jdk8
-RUN yum -y install java-1.8.0-openjdk java-1.8.0-openjdk-devel \
- && echo "export JAVA_8_HOME=/usr/lib/jvm/java-1.8.0-openjdk" >> /etc/profile \
- && source /etc/profile
-
-ENV JAVA_HOME=/usr/lib/jvm/java-11-openjdk
-ENV JAVA_8_HOME=/usr/lib/jvm/java-1.8.0-openjdk
-ENV USE_LOCAL_K8S=false
+ENV IS_LOCAL_MODE=false
ENV K8S_MASTER_URL=http://127.0.0.1:8000/
ENV HTTP_SERVER_PORT=8080
diff --git a/geaflow-kubernetes-operator/docker/bin/start-process.sh b/geaflow-kubernetes-operator/docker/bin/start-process.sh
index fb915efe2..110ad29f7 100755
--- a/geaflow-kubernetes-operator/docker/bin/start-process.sh
+++ b/geaflow-kubernetes-operator/docker/bin/start-process.sh
@@ -18,39 +18,21 @@
# under the License.
#
-set -x
-
-source /etc/profile
+set -euo pipefail
# prepare path
BASE_LOG_DIR=/tmp/logs
GEAFLOW_LOG_DIR=$BASE_LOG_DIR/geaflow
-mkdir -p $BASE_LOG_DIR
-mkdir -p $GEAFLOW_LOG_DIR
-if [[ ! -L $GEAFLOW_HOME/logs ]]; then
- ln -s $BASE_LOG_DIR $GEAFLOW_HOME/logs
+mkdir -p "$GEAFLOW_LOG_DIR"
+if [[ ! -e "$GEAFLOW_HOME/logs" && ! -L "$GEAFLOW_HOME/logs" ]]; then
+ ln -s "$BASE_LOG_DIR" "$GEAFLOW_HOME/logs"
fi
-function startGeaflowOperator() {
- K8S_SERVICE_ACCOUNT=${K8S_SERVICE_ACCOUNT:-"default"}
- PID_FILE=$GEAFLOW_HOME/geaflow.pid
- ps -p $(cat $PID_FILE 2>/dev/null) &> /dev/null && echo "geaflow-operator has been started" || {
- cd $GEAFLOW_HOME
- nohup $JAVA_HOME/bin/java -Xmx2048m -cp $GEAFLOW_HOME/boot/geaflow-kubernetes-operator-bootstrap.jar \
- org.springframework.boot.loader.JarLauncher --java_8_home=$JAVA_8_HOME \
- --is_local_mode=$IS_LOCAL_MODE --k8s_master_url=$K8S_MASTER_URL \
- --server.port=$HTTP_SERVER_PORT \
- > $GEAFLOW_LOG_DIR/stdout.log 2>$GEAFLOW_LOG_DIR/stderr.log &
- echo "start geaflow success"
- echo $! >$PID_FILE
- }
-}
-
-# start geaflow-operator
-startGeaflowOperator || exit 1
-
-# work dir
-cd $GEAFLOW_HOME && echo "work directory: $(pwd)"
-
-# docker entrypoint
-tail -f /dev/null
+# Keep the JVM as PID 1 so startup failures and termination reach the container runtime.
+cd "$GEAFLOW_HOME"
+exec "$JAVA_HOME/bin/java" -Xmx2048m \
+ -cp "$GEAFLOW_HOME/boot/geaflow-kubernetes-operator-bootstrap.jar" \
+ org.springframework.boot.loader.JarLauncher \
+ --is_local_mode="${IS_LOCAL_MODE:-false}" \
+ --k8s_master_url="${K8S_MASTER_URL:-http://127.0.0.1:8000/}" \
+ --server.port="${HTTP_SERVER_PORT:-8080}" "$@"
diff --git a/geaflow-kubernetes-operator/geaflow-kubernetes-operator-bootstrap/src/main/resources/application.properties b/geaflow-kubernetes-operator/geaflow-kubernetes-operator-bootstrap/src/main/resources/application.properties
index ad0b812b5..ce90df138 100644
--- a/geaflow-kubernetes-operator/geaflow-kubernetes-operator-bootstrap/src/main/resources/application.properties
+++ b/geaflow-kubernetes-operator/geaflow-kubernetes-operator-bootstrap/src/main/resources/application.properties
@@ -29,4 +29,9 @@ k8s_master_url=http://127.0.0.1:8001/
k8s_namespace=default
k8s_client_cert_data=
k8s_client_key_data=
-k8s_ca_cert_data=
\ No newline at end of file
+k8s_ca_cert_data=
+
+# Required: use a high-entropy token from a Secret, never a command-line argument.
+# Missing/invalid administrator credentials prevent the web application from starting.
+geaflow.operator.web.security.admin-token=${GEAFLOW_OPERATOR_ADMIN_TOKEN:}
+geaflow.operator.web.security.viewer-token=${GEAFLOW_OPERATOR_VIEWER_TOKEN:}
diff --git a/geaflow-kubernetes-operator/geaflow-kubernetes-operator-core/src/main/java/org/apache/geaflow/kubernetes/operator/core/util/KubernetesUtil.java b/geaflow-kubernetes-operator/geaflow-kubernetes-operator-core/src/main/java/org/apache/geaflow/kubernetes/operator/core/util/KubernetesUtil.java
index 5c2b57347..55401bc14 100644
--- a/geaflow-kubernetes-operator/geaflow-kubernetes-operator-core/src/main/java/org/apache/geaflow/kubernetes/operator/core/util/KubernetesUtil.java
+++ b/geaflow-kubernetes-operator/geaflow-kubernetes-operator-core/src/main/java/org/apache/geaflow/kubernetes/operator/core/util/KubernetesUtil.java
@@ -19,6 +19,14 @@
package org.apache.geaflow.kubernetes.operator.core.util;
+import io.fabric8.kubernetes.api.model.ConfigMap;
+import io.fabric8.kubernetes.api.model.ContainerStateRunning;
+import io.fabric8.kubernetes.api.model.ContainerStateWaiting;
+import io.fabric8.kubernetes.api.model.ContainerStatus;
+import io.fabric8.kubernetes.api.model.ListOptions;
+import io.fabric8.kubernetes.api.model.Pod;
+import io.fabric8.kubernetes.api.model.PodList;
+import io.fabric8.kubernetes.api.model.Service;
import io.fabric8.kubernetes.api.model.apps.Deployment;
import io.fabric8.kubernetes.client.Config;
import io.fabric8.kubernetes.client.KubernetesClient;
diff --git a/geaflow-kubernetes-operator/geaflow-kubernetes-operator-web/pom.xml b/geaflow-kubernetes-operator/geaflow-kubernetes-operator-web/pom.xml
index 0aa34bf29..5d76a42d3 100644
--- a/geaflow-kubernetes-operator/geaflow-kubernetes-operator-web/pom.xml
+++ b/geaflow-kubernetes-operator/geaflow-kubernetes-operator-web/pom.xml
@@ -60,6 +60,35 @@
org.springframework.bootspring-boot-starter-log4j2
+
+
+ org.springframework.boot
+ spring-boot-starter-security
+
+
+ org.springframework.boot
+ spring-boot-starter-logging
+
+
+
+
+
+ org.springframework.boot
+ spring-boot-starter-test
+
+
+ org.springframework.boot
+ spring-boot-starter-logging
+
+
+ test
+
+
+
+ junit
+ junit
+ test
+
diff --git a/geaflow-kubernetes-operator/geaflow-kubernetes-operator-web/src/main/java/org/apache/geaflow/kubernetes/operator/web/controller/api/JobController.java b/geaflow-kubernetes-operator/geaflow-kubernetes-operator-web/src/main/java/org/apache/geaflow/kubernetes/operator/web/controller/api/JobController.java
index ed1a2e1ce..503ea4d18 100644
--- a/geaflow-kubernetes-operator/geaflow-kubernetes-operator-web/src/main/java/org/apache/geaflow/kubernetes/operator/web/controller/api/JobController.java
+++ b/geaflow-kubernetes-operator/geaflow-kubernetes-operator-web/src/main/java/org/apache/geaflow/kubernetes/operator/web/controller/api/JobController.java
@@ -26,6 +26,13 @@
import org.apache.geaflow.kubernetes.operator.web.converter.GeaflowJobConverter;
import org.apache.geaflow.kubernetes.operator.web.view.GeaflowJobView;
import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.web.bind.annotation.DeleteMapping;
+import org.springframework.web.bind.annotation.GetMapping;
+import org.springframework.web.bind.annotation.PathVariable;
+import org.springframework.web.bind.annotation.PostMapping;
+import org.springframework.web.bind.annotation.RequestBody;
+import org.springframework.web.bind.annotation.RequestMapping;
+import org.springframework.web.bind.annotation.RestController;
@RestController
@RequestMapping("/jobs")
diff --git a/geaflow-kubernetes-operator/geaflow-kubernetes-operator-web/src/main/java/org/apache/geaflow/kubernetes/operator/web/security/BearerTokenFilter.java b/geaflow-kubernetes-operator/geaflow-kubernetes-operator-web/src/main/java/org/apache/geaflow/kubernetes/operator/web/security/BearerTokenFilter.java
new file mode 100644
index 000000000..231cbbc39
--- /dev/null
+++ b/geaflow-kubernetes-operator/geaflow-kubernetes-operator-web/src/main/java/org/apache/geaflow/kubernetes/operator/web/security/BearerTokenFilter.java
@@ -0,0 +1,63 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.geaflow.kubernetes.operator.web.security;
+
+import java.io.IOException;
+import java.util.Enumeration;
+import javax.servlet.FilterChain;
+import javax.servlet.ServletException;
+import javax.servlet.http.HttpServletRequest;
+import javax.servlet.http.HttpServletResponse;
+import org.springframework.http.HttpHeaders;
+import org.springframework.security.core.Authentication;
+import org.springframework.security.core.context.SecurityContextHolder;
+import org.springframework.web.filter.OncePerRequestFilter;
+
+final class BearerTokenFilter extends OncePerRequestFilter {
+
+ private final OperatorTokens tokens;
+
+ BearerTokenFilter(OperatorTokens tokens) {
+ this.tokens = tokens;
+ }
+
+ @Override
+ protected void doFilterInternal(HttpServletRequest request, HttpServletResponse response,
+ FilterChain chain) throws ServletException, IOException {
+ Enumeration headers = request.getHeaders(HttpHeaders.AUTHORIZATION);
+ if (headers == null || !headers.hasMoreElements()) {
+ chain.doFilter(request, response);
+ return;
+ }
+ String header = headers.nextElement();
+ if (headers.hasMoreElements() || header.length() < 8 || !header.regionMatches(true, 0, "Bearer ", 0, 7)) {
+ SecurityResponse.unauthorized(response);
+ return;
+ }
+ Authentication authentication = tokens.authenticate(header.substring(7));
+ if (authentication == null) {
+ SecurityResponse.unauthorized(response);
+ return;
+ }
+ // Never keep the raw credential in the principal, session, request attributes or logs.
+ SecurityContextHolder.getContext().setAuthentication(authentication);
+ chain.doFilter(request, response);
+ }
+}
diff --git a/geaflow-kubernetes-operator/geaflow-kubernetes-operator-web/src/main/java/org/apache/geaflow/kubernetes/operator/web/security/OperatorTokens.java b/geaflow-kubernetes-operator/geaflow-kubernetes-operator-web/src/main/java/org/apache/geaflow/kubernetes/operator/web/security/OperatorTokens.java
new file mode 100644
index 000000000..1fae6fc4c
--- /dev/null
+++ b/geaflow-kubernetes-operator/geaflow-kubernetes-operator-web/src/main/java/org/apache/geaflow/kubernetes/operator/web/security/OperatorTokens.java
@@ -0,0 +1,86 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.geaflow.kubernetes.operator.web.security;
+
+import java.nio.charset.StandardCharsets;
+import java.security.MessageDigest;
+import java.security.NoSuchAlgorithmException;
+import java.util.Collections;
+import org.springframework.security.authentication.UsernamePasswordAuthenticationToken;
+import org.springframework.security.core.Authentication;
+import org.springframework.security.core.authority.SimpleGrantedAuthority;
+
+/** Deployment-scoped credentials. Job submitters are trusted to execute code in the cluster. */
+public final class OperatorTokens {
+
+ private final byte[] adminDigest;
+ private final byte[] viewerDigest;
+
+ public OperatorTokens(String adminToken, String viewerToken) {
+ validate(adminToken, "admin-token");
+ adminDigest = digest(adminToken);
+ if (viewerToken == null || viewerToken.isEmpty()) {
+ viewerDigest = null;
+ } else {
+ validate(viewerToken, "viewer-token");
+ viewerDigest = digest(viewerToken);
+ if (MessageDigest.isEqual(adminDigest, viewerDigest)) {
+ throw new IllegalArgumentException("Operator admin and viewer tokens must be different");
+ }
+ }
+ }
+
+ public Authentication authenticate(String token) {
+ if (token == null || token.length() > 512) {
+ return null;
+ }
+ byte[] candidate = digest(token);
+ boolean admin = MessageDigest.isEqual(adminDigest, candidate);
+ boolean viewer = viewerDigest != null && MessageDigest.isEqual(viewerDigest, candidate);
+ if (admin) {
+ return authentication("operator-admin", "ROLE_ADMIN");
+ }
+ if (viewer) {
+ return authentication("operator-viewer", "ROLE_VIEWER");
+ }
+ return null;
+ }
+
+ private static Authentication authentication(String name, String role) {
+ return new UsernamePasswordAuthenticationToken(name, null,
+ Collections.singletonList(new SimpleGrantedAuthority(role)));
+ }
+
+ private static void validate(String token, String property) {
+ // RFC 6750 b64token syntax. Require high-entropy, independently generated tokens.
+ if (token == null || !token.matches("[A-Za-z0-9+/_~.\\-]{32,512}={0,2}") || token.length() > 512) {
+ throw new IllegalArgumentException("Configure geaflow.operator.web.security." + property
+ + " with a random bearer token of 32 to 512 characters");
+ }
+ }
+
+ private static byte[] digest(String token) {
+ try {
+ return MessageDigest.getInstance("SHA-256").digest(token.getBytes(StandardCharsets.UTF_8));
+ } catch (NoSuchAlgorithmException e) {
+ throw new IllegalStateException("SHA-256 is unavailable", e);
+ }
+ }
+}
diff --git a/geaflow-kubernetes-operator/geaflow-kubernetes-operator-web/src/main/java/org/apache/geaflow/kubernetes/operator/web/security/SecurityResponse.java b/geaflow-kubernetes-operator/geaflow-kubernetes-operator-web/src/main/java/org/apache/geaflow/kubernetes/operator/web/security/SecurityResponse.java
new file mode 100644
index 000000000..862e0be24
--- /dev/null
+++ b/geaflow-kubernetes-operator/geaflow-kubernetes-operator-web/src/main/java/org/apache/geaflow/kubernetes/operator/web/security/SecurityResponse.java
@@ -0,0 +1,53 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.geaflow.kubernetes.operator.web.security;
+
+import java.io.IOException;
+import java.nio.charset.StandardCharsets;
+import javax.servlet.http.HttpServletResponse;
+import org.springframework.http.HttpHeaders;
+import org.springframework.http.MediaType;
+
+final class SecurityResponse {
+
+ private SecurityResponse() {
+ }
+
+ static void badRequest(HttpServletResponse response) throws IOException {
+ write(response, HttpServletResponse.SC_BAD_REQUEST, "Invalid request");
+ }
+
+ static void unauthorized(HttpServletResponse response) throws IOException {
+ response.setHeader(HttpHeaders.WWW_AUTHENTICATE, "Bearer realm=\"geaflow-operator\"");
+ write(response, HttpServletResponse.SC_UNAUTHORIZED, "Authentication required");
+ }
+
+ static void forbidden(HttpServletResponse response) throws IOException {
+ write(response, HttpServletResponse.SC_FORBIDDEN, "Access denied");
+ }
+
+ private static void write(HttpServletResponse response, int status, String message) throws IOException {
+ response.setStatus(status);
+ response.setContentType(MediaType.APPLICATION_JSON_VALUE);
+ response.setCharacterEncoding(StandardCharsets.UTF_8.name());
+ response.setHeader(HttpHeaders.CACHE_CONTROL, "no-store");
+ response.getWriter().write("{\"success\":false,\"message\":\"" + message + "\"}");
+ }
+}
diff --git a/geaflow-kubernetes-operator/geaflow-kubernetes-operator-web/src/main/java/org/apache/geaflow/kubernetes/operator/web/security/WebSecurityConfiguration.java b/geaflow-kubernetes-operator/geaflow-kubernetes-operator-web/src/main/java/org/apache/geaflow/kubernetes/operator/web/security/WebSecurityConfiguration.java
new file mode 100644
index 000000000..f03515a59
--- /dev/null
+++ b/geaflow-kubernetes-operator/geaflow-kubernetes-operator-web/src/main/java/org/apache/geaflow/kubernetes/operator/web/security/WebSecurityConfiguration.java
@@ -0,0 +1,87 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.geaflow.kubernetes.operator.web.security;
+
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.http.HttpMethod;
+import org.springframework.security.authentication.AuthenticationManager;
+import org.springframework.security.authentication.BadCredentialsException;
+import org.springframework.security.config.annotation.web.builders.HttpSecurity;
+import org.springframework.security.config.annotation.web.configuration.EnableWebSecurity;
+import org.springframework.security.config.annotation.web.configuration.WebSecurityCustomizer;
+import org.springframework.security.config.http.SessionCreationPolicy;
+import org.springframework.security.web.SecurityFilterChain;
+import org.springframework.security.web.authentication.AnonymousAuthenticationFilter;
+
+@Configuration
+@EnableWebSecurity
+public class WebSecurityConfiguration {
+
+ @Bean
+ public OperatorTokens operatorTokens(
+ @Value("${geaflow.operator.web.security.admin-token:}") String adminToken,
+ @Value("${geaflow.operator.web.security.viewer-token:}") String viewerToken) {
+ return new OperatorTokens(adminToken, viewerToken);
+ }
+
+ @Bean
+ public WebSecurityCustomizer operatorFirewallErrors() {
+ // Keep StrictHttpFirewall's path checks and return a generic response without an error dispatch.
+ return web -> web.requestRejectedHandler((request, response, error) -> SecurityResponse.badRequest(response));
+ }
+
+ @Bean
+ public AuthenticationManager operatorAuthenticationManager() {
+ // Disable fallback accounts, including Spring Boot's generated default user.
+ return authentication -> {
+ throw new BadCredentialsException("Bearer token authentication required");
+ };
+ }
+
+ @Bean
+ public SecurityFilterChain operatorSecurityFilterChain(HttpSecurity http, OperatorTokens tokens,
+ AuthenticationManager authenticationManager) throws Exception {
+ // Only explicit Authorization headers authenticate requests. No cookies, Basic auth,
+ // form login or sessions are accepted, so browsers cannot send credentials implicitly.
+ http.authenticationManager(authenticationManager)
+ .csrf().disable()
+ .cors().disable()
+ .httpBasic().disable()
+ .formLogin().disable()
+ .logout().disable()
+ .requestCache().disable()
+ .sessionManagement().sessionCreationPolicy(SessionCreationPolicy.STATELESS)
+ .and().exceptionHandling()
+ .authenticationEntryPoint((request, response, error) -> SecurityResponse.unauthorized(response))
+ .accessDeniedHandler((request, response, error) -> SecurityResponse.forbidden(response))
+ .and().authorizeRequests()
+ .antMatchers(HttpMethod.GET, "/api", "/api/**").hasAnyRole("ADMIN", "VIEWER")
+ .antMatchers(HttpMethod.HEAD, "/api", "/api/**").hasAnyRole("ADMIN", "VIEWER")
+ .antMatchers("/api", "/api/**").hasRole("ADMIN")
+ .antMatchers(HttpMethod.GET, "/health", "/", "/index.html", "/*.ico", "/*.svg",
+ "/*.js", "/*.css", "/*.png", "/*.jpg", "/*.jpeg", "/*.gif", "/*.webp",
+ "/*.woff", "/*.woff2", "/*.ttf", "/*.eot", "/scripts/**", "/static/**", "/assets/**", "/icons/**").permitAll()
+ .anyRequest().denyAll()
+ .and().addFilterBefore(new BearerTokenFilter(tokens), AnonymousAuthenticationFilter.class);
+ return http.build();
+ }
+}
diff --git a/geaflow-kubernetes-operator/geaflow-kubernetes-operator-web/src/test/java/org/apache/geaflow/kubernetes/operator/web/security/OperatorTokensTest.java b/geaflow-kubernetes-operator/geaflow-kubernetes-operator-web/src/test/java/org/apache/geaflow/kubernetes/operator/web/security/OperatorTokensTest.java
new file mode 100644
index 000000000..69fefec6e
--- /dev/null
+++ b/geaflow-kubernetes-operator/geaflow-kubernetes-operator-web/src/test/java/org/apache/geaflow/kubernetes/operator/web/security/OperatorTokensTest.java
@@ -0,0 +1,76 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.geaflow.kubernetes.operator.web.security;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertThrows;
+
+import java.util.Arrays;
+import org.junit.Test;
+import org.springframework.boot.test.context.runner.WebApplicationContextRunner;
+import org.springframework.security.core.Authentication;
+
+public class OperatorTokensTest {
+
+ private static final String ADMIN = "0123456789abcdef0123456789abcdef";
+ private static final String VIEWER = "abcdef0123456789abcdef0123456789";
+
+ @Test
+ public void rejectsMissingWeakMalformedAndIdenticalCredentials() {
+ for (String invalid : Arrays.asList(null, "", " ", "short", ADMIN + " ", "Bearer " + ADMIN,
+ ADMIN + "\n", "!" + ADMIN, "a".repeat(513))) {
+ assertThrows(IllegalArgumentException.class, () -> new OperatorTokens(invalid, ""));
+ }
+ assertThrows(IllegalArgumentException.class, () -> new OperatorTokens(ADMIN, "short"));
+ assertThrows(IllegalArgumentException.class, () -> new OperatorTokens(ADMIN, ADMIN));
+ }
+
+ @Test
+ public void assignsSeparateRolesAndErasesCredentials() {
+ OperatorTokens tokens = new OperatorTokens(ADMIN, VIEWER);
+ Authentication admin = tokens.authenticate(ADMIN);
+ assertEquals("ROLE_ADMIN", admin.getAuthorities().iterator().next().getAuthority());
+ assertNull(admin.getCredentials());
+ assertFalse(admin.toString().contains(ADMIN));
+ assertEquals("ROLE_VIEWER", tokens.authenticate(VIEWER).getAuthorities().iterator().next().getAuthority());
+ assertNull(tokens.authenticate(ADMIN + "x"));
+ assertNull(tokens.authenticate(ADMIN.toUpperCase()));
+ assertNull(tokens.authenticate(null));
+ assertNull(tokens.authenticate("a".repeat(513)));
+ assertNull(new OperatorTokens(ADMIN, "").authenticate(VIEWER));
+ }
+
+ @Test
+ public void failsStartupWithoutValidAdministratorToken() {
+ new WebApplicationContextRunner().withUserConfiguration(WebSecurityConfiguration.class)
+ .run(context -> {
+ assertThat(context).hasFailed();
+ Throwable cause = context.getStartupFailure();
+ while (cause.getCause() != null) {
+ cause = cause.getCause();
+ }
+ assertEquals("Configure geaflow.operator.web.security.admin-token"
+ + " with a random bearer token of 32 to 512 characters", cause.getMessage());
+ });
+ }
+}
diff --git a/geaflow-kubernetes-operator/geaflow-kubernetes-operator-web/src/test/java/org/apache/geaflow/kubernetes/operator/web/security/WebSecurityTest.java b/geaflow-kubernetes-operator/geaflow-kubernetes-operator-web/src/test/java/org/apache/geaflow/kubernetes/operator/web/security/WebSecurityTest.java
new file mode 100644
index 000000000..f41630c3b
--- /dev/null
+++ b/geaflow-kubernetes-operator/geaflow-kubernetes-operator-web/src/test/java/org/apache/geaflow/kubernetes/operator/web/security/WebSecurityTest.java
@@ -0,0 +1,227 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.geaflow.kubernetes.operator.web.security;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNull;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.reset;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoInteractions;
+import static org.mockito.Mockito.when;
+import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.delete;
+import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.get;
+import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.head;
+import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.options;
+import static org.springframework.test.web.servlet.request.MockMvcRequestBuilders.post;
+import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.content;
+import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.header;
+import static org.springframework.test.web.servlet.result.MockMvcResultMatchers.status;
+
+import java.net.URI;
+import java.util.Collections;
+import javax.servlet.Filter;
+import javax.servlet.http.Cookie;
+import org.apache.geaflow.kubernetes.operator.core.job.GeaflowJobApiService;
+import org.apache.geaflow.kubernetes.operator.core.job.JobCRCache;
+import org.apache.geaflow.kubernetes.operator.core.model.customresource.GeaflowJob;
+import org.apache.geaflow.kubernetes.operator.web.controller.HealthController;
+import org.apache.geaflow.kubernetes.operator.web.controller.api.JobController;
+import org.apache.geaflow.kubernetes.operator.web.mvc.ApiFilter;
+import org.apache.geaflow.kubernetes.operator.web.mvc.WebConfig;
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.ArgumentCaptor;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.Import;
+import org.springframework.http.HttpHeaders;
+import org.springframework.http.MediaType;
+import org.springframework.mock.web.MockHttpSession;
+import org.springframework.security.core.context.SecurityContextImpl;
+import org.springframework.security.web.context.HttpSessionSecurityContextRepository;
+import org.springframework.test.context.ContextConfiguration;
+import org.springframework.test.context.TestPropertySource;
+import org.springframework.test.context.junit4.SpringRunner;
+import org.springframework.test.context.web.WebAppConfiguration;
+import org.springframework.test.web.servlet.MockMvc;
+import org.springframework.test.web.servlet.request.MockHttpServletRequestBuilder;
+import org.springframework.test.web.servlet.setup.MockMvcBuilders;
+import org.springframework.web.context.WebApplicationContext;
+import org.springframework.web.servlet.config.annotation.EnableWebMvc;
+
+@RunWith(SpringRunner.class)
+@WebAppConfiguration
+@ContextConfiguration(classes = WebSecurityTest.TestConfiguration.class)
+@TestPropertySource(properties = {
+ "geaflow.operator.web.security.admin-token=" + WebSecurityTest.ADMIN,
+ "geaflow.operator.web.security.viewer-token=" + WebSecurityTest.VIEWER
+})
+public class WebSecurityTest {
+
+ static final String ADMIN = "0123456789abcdef0123456789abcdef";
+ static final String VIEWER = "abcdef0123456789abcdef0123456789";
+ private static final String JOB = "{\"name\":\"security-regression\",\"spec\":{"
+ + "\"image\":\"example.invalid/trusted-job:1\",\"serviceAccount\":\"geaflow\","
+ + "\"entryClass\":\"org.example.Job\"}}";
+
+ @Autowired
+ private WebApplicationContext context;
+
+ @Autowired
+ @Qualifier("springSecurityFilterChain")
+ private Filter securityFilter;
+
+ @Autowired
+ private GeaflowJobApiService service;
+
+ private MockMvc mvc;
+
+ @Before
+ public void setup() {
+ reset(service);
+ when(service.queryJobs()).thenReturn(Collections.emptyList());
+ mvc = MockMvcBuilders.webAppContextSetup(context)
+ .addFilters(securityFilter, new ApiFilter()).build();
+ }
+
+ @Test
+ public void anonymousRequestsCannotReachBusinessServices() throws Exception {
+ for (MockHttpServletRequestBuilder request : new MockHttpServletRequestBuilder[] {
+ get("/api/jobs"), head("/api/jobs"), post("/api/jobs").contentType(MediaType.APPLICATION_JSON).content(JOB),
+ post("/api/jobs").contentType(MediaType.APPLICATION_JSON).content("{"),
+ delete("/api/jobs/security-regression"), get("/api/overview"), get("/api/future"), get("/api/future.js"), options("/api/jobs")}) {
+ mvc.perform(request).andExpect(status().isUnauthorized())
+ .andExpect(header().string(HttpHeaders.WWW_AUTHENTICATE, "Bearer realm=\"geaflow-operator\""))
+ .andExpect(header().string(HttpHeaders.CACHE_CONTROL, "no-store"));
+ }
+ verifyNoInteractions(service);
+ }
+
+ @Test
+ public void rejectsInvalidAndAmbiguousAuthorization() throws Exception {
+ for (String value : new String[] {"", "Basic " + ADMIN, "Bearer", "Bearer bad", "Bearer " + ADMIN + " ",
+ "Bearer " + ADMIN + ",Bearer " + VIEWER, "Bearer " + ADMIN + "x"}) {
+ mvc.perform(get("/api/jobs").header(HttpHeaders.AUTHORIZATION, value))
+ .andExpect(status().isUnauthorized());
+ }
+ mvc.perform(get("/api/jobs").header(HttpHeaders.AUTHORIZATION, "Bearer " + ADMIN, "Bearer " + VIEWER))
+ .andExpect(status().isUnauthorized());
+ verifyNoInteractions(service);
+ }
+
+ @Test
+ public void ignoresQueryCookiesForwardedIdentityAndExistingSessions() throws Exception {
+ MockHttpSession session = new MockHttpSession();
+ session.setAttribute(HttpSessionSecurityContextRepository.SPRING_SECURITY_CONTEXT_KEY,
+ new SecurityContextImpl(new OperatorTokens(ADMIN, VIEWER).authenticate(ADMIN)));
+ mvc.perform(get("/api/jobs").param("access_token", ADMIN)
+ .cookie(new Cookie("Authorization", "Bearer " + ADMIN))
+ .header("X-Forwarded-User", "admin").header("X-Auth-Request-User", "admin").session(session))
+ .andExpect(status().isUnauthorized());
+ verifyNoInteractions(service);
+ }
+
+ @Test
+ public void viewerCanReadButCannotCreateDeleteOrUseOtherMethods() throws Exception {
+ mvc.perform(get("/api/jobs").header(HttpHeaders.AUTHORIZATION, "Bearer " + VIEWER))
+ .andExpect(status().isOk());
+ mvc.perform(head("/api/jobs").header(HttpHeaders.AUTHORIZATION, "Bearer " + VIEWER))
+ .andExpect(status().isOk());
+ reset(service);
+ for (MockHttpServletRequestBuilder request : new MockHttpServletRequestBuilder[] {
+ post("/api/jobs").contentType(MediaType.APPLICATION_JSON).content(JOB),
+ delete("/api/jobs/security-regression"), options("/api/jobs"), post("/api/future")}) {
+ mvc.perform(request.header(HttpHeaders.AUTHORIZATION, "Bearer " + VIEWER))
+ .andExpect(status().isForbidden()).andExpect(content().json("{\"success\":false,\"message\":\"Access denied\"}"));
+ }
+ verifyNoInteractions(service);
+ }
+
+ @Test
+ public void administratorCanSubmitAndDeleteJobsWithoutCreatingSession() throws Exception {
+ mvc.perform(post("/api/jobs").header(HttpHeaders.AUTHORIZATION, "Bearer " + ADMIN)
+ .contentType(MediaType.APPLICATION_JSON).content(JOB))
+ .andExpect(status().isOk()).andExpect(result -> assertNull(result.getRequest().getSession(false)));
+ ArgumentCaptor job = ArgumentCaptor.forClass(GeaflowJob.class);
+ verify(service).createJob(job.capture());
+ assertEquals("example.invalid/trusted-job:1", job.getValue().getSpec().getImage());
+ assertEquals("geaflow", job.getValue().getSpec().getServiceAccount());
+ mvc.perform(delete("/api/jobs/security-regression").header(HttpHeaders.AUTHORIZATION, "Bearer " + ADMIN))
+ .andExpect(status().isOk());
+ verify(service).deleteJob("security-regression");
+ mvc.perform(get("/api/jobs")).andExpect(status().isUnauthorized());
+ }
+
+ @Test
+ public void acceptsCaseInsensitiveSchemeAndContextPath() throws Exception {
+ mvc.perform(get("/operator/api/jobs").contextPath("/operator")
+ .header(HttpHeaders.AUTHORIZATION, "bearer " + VIEWER))
+ .andExpect(status().isOk());
+ }
+
+ @Test
+ public void exposesHealthButDoesNotExemptOtherHealthPathsOrUnprefixedControllers() throws Exception {
+ mvc.perform(get("/health")).andExpect(status().isOk());
+ for (String path : new String[] {"/health/jobs", "/jobs", "/overview", "/api/jobs/", "/api/future"}) {
+ mvc.perform(get(path)).andExpect(status().isUnauthorized());
+ }
+ mvc.perform(get("/jobs").header(HttpHeaders.AUTHORIZATION, "Bearer " + ADMIN))
+ .andExpect(status().isForbidden());
+ verifyNoInteractions(service);
+ }
+
+ @Test
+ public void rejectsNonCanonicalPathsBeforeControllerDispatch() throws Exception {
+ for (String path : new String[] {"/api//jobs", "/api/./jobs", "/api/jobs;x=1"}) {
+ mvc.perform(get(URI.create(path)).header(HttpHeaders.AUTHORIZATION, "Bearer " + ADMIN))
+ .andExpect(result -> assertEquals(path, 400, result.getResponse().getStatus()));
+ }
+ verifyNoInteractions(service);
+ }
+
+ @Test
+ public void corsPreflightDoesNotGrantAnonymousAccess() throws Exception {
+ mvc.perform(options("/api/jobs").header("Origin", "https://untrusted.invalid")
+ .header("Access-Control-Request-Method", "POST")
+ .header("Access-Control-Request-Headers", "authorization"))
+ .andExpect(status().isUnauthorized()).andExpect(header().doesNotExist("Access-Control-Allow-Origin"));
+ verifyNoInteractions(service);
+ }
+
+ @Configuration
+ @EnableWebMvc
+ @Import({WebSecurityConfiguration.class, WebConfig.class, JobController.class, HealthController.class})
+ static class TestConfiguration {
+
+ @Bean
+ GeaflowJobApiService geaflowJobApiService() {
+ return mock(GeaflowJobApiService.class);
+ }
+
+ @Bean
+ JobCRCache jobCRCache() {
+ return mock(JobCRCache.class);
+ }
+ }
+}
diff --git a/geaflow-kubernetes-operator/geaflow-kubernetes-operator-web/web-dashboard/src/app.tsx b/geaflow-kubernetes-operator/geaflow-kubernetes-operator-web/web-dashboard/src/app.tsx
index acfaf0008..546bfb698 100644
--- a/geaflow-kubernetes-operator/geaflow-kubernetes-operator-web/web-dashboard/src/app.tsx
+++ b/geaflow-kubernetes-operator/geaflow-kubernetes-operator-web/web-dashboard/src/app.tsx
@@ -18,6 +18,7 @@
*/
import Footer from '@/components/Footer';
+import OperatorAccess from '@/components/OperatorAccess';
import {Question, SelectLang} from '@/components/RightContent';
import {LinkOutlined} from '@ant-design/icons';
import type {Settings as LayoutSettings} from '@ant-design/pro-components';
@@ -43,7 +44,7 @@ export async function getInitialState(): Promise<{
}> {
const fetchUserInfo = async () => {
return {
- name: 'Admin',
+ name: 'Operator',
avatar: 'https://gw.alipayobjects.com/zos/antfincdn/XAosXuNZyF/BiazfanxmamNRoxxVxka.png',
};
};
@@ -109,7 +110,7 @@ export const layout: RunTimeLayoutConfig = ({ initialState, setInitialState }) =
// if (initialState?.loading) return ;
return (
<>
- {children}
+ {children}
{isDev ? = ({ children }) => {
+ const [authenticated, setAuthenticated] = useState(!!getOperatorToken());
+ const [loading, setLoading] = useState(false);
+ const [error, setError] = useState('');
+ const [form] = Form.useForm();
+
+ useEffect(() => onOperatorAuthChanged(() => setAuthenticated(!!getOperatorToken())), []);
+
+ const connect = async ({ token }: { token: string }) => {
+ setLoading(true);
+ setError('');
+ try {
+ const response = await fetch(operatorApiUrl('/api/jobs'), {
+ headers: { Authorization: `Bearer ${token}` },
+ credentials: 'omit',
+ redirect: 'error',
+ cache: 'no-store',
+ });
+ if (!response.ok) {
+ throw new Error('Access failed. Check the token and operator connection.');
+ }
+ const body = await response.json();
+ if (!body.success) {
+ throw new Error('The operator could not verify access.');
+ }
+ form.resetFields();
+ setOperatorToken(token);
+ } catch {
+ setError('Access failed. Check the token and operator connection.');
+ } finally {
+ setLoading(false);
+ }
+ };
+
+ if (authenticated) {
+ return <>
+
+ {children}
+ ;
+ }
+
+ return
+
Enter your administrator or read-only token. Access is cleared when this page reloads.
+ {error && }
+
+ ;
+};
+
+export default OperatorAccess;
diff --git a/geaflow-kubernetes-operator/geaflow-kubernetes-operator-web/web-dashboard/src/operatorAuth.test.ts b/geaflow-kubernetes-operator/geaflow-kubernetes-operator-web/web-dashboard/src/operatorAuth.test.ts
new file mode 100644
index 000000000..241f348d1
--- /dev/null
+++ b/geaflow-kubernetes-operator/geaflow-kubernetes-operator-web/web-dashboard/src/operatorAuth.test.ts
@@ -0,0 +1,59 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+import { getOperatorToken, onOperatorAuthChanged, operatorApiUrl, operatorAuthHeaders, setOperatorToken } from './operatorAuth';
+
+beforeEach(() => {
+ setOperatorToken();
+ window.history.replaceState({}, '', '/');
+});
+
+test('sends tokens only to the same operator API, including proxy prefixes', () => {
+ setOperatorToken('test-token');
+ expect(operatorAuthHeaders(operatorApiUrl('/api/jobs'))).toEqual({ Authorization: 'Bearer test-token' });
+ for (const target of ['https://untrusted.invalid/api/jobs', '/api-other/jobs', '/scripts/test.js']) {
+ expect(operatorAuthHeaders(target)).toEqual({});
+ }
+ window.history.replaceState({}, '', '/operator/index.html');
+ expect(operatorApiUrl('/api/jobs')).toBe(`${window.location.origin}/operator/api/jobs`);
+ expect(operatorAuthHeaders('/api/jobs')).toEqual({});
+ expect(operatorAuthHeaders(operatorApiUrl('/api/jobs'))).toEqual({ Authorization: 'Bearer test-token' });
+ window.history.replaceState({}, '', '/operator/');
+ expect(operatorApiUrl('/api/jobs')).toBe(`${window.location.origin}/operator/api/jobs`);
+});
+
+test('clears credentials and notifies mounted consumers', () => {
+ const listener = jest.fn();
+ const unsubscribe = onOperatorAuthChanged(listener);
+ setOperatorToken('test-token');
+ expect(getOperatorToken()).toBe('test-token');
+ setOperatorToken();
+ expect(getOperatorToken()).toBeUndefined();
+ expect(operatorAuthHeaders(operatorApiUrl('/api/jobs'))).toEqual({});
+ expect(listener).toHaveBeenCalledTimes(2);
+ unsubscribe();
+ setOperatorToken();
+ expect(listener).toHaveBeenCalledTimes(2);
+});
+
+test('rejects paths outside the operator API', () => {
+ for (const path of ['https://untrusted.invalid/api/jobs', '//untrusted.invalid/api/jobs', '/health', '']) {
+ expect(() => operatorApiUrl(path)).toThrow('Invalid operator API path');
+ }
+});
diff --git a/geaflow-kubernetes-operator/geaflow-kubernetes-operator-web/web-dashboard/src/operatorAuth.ts b/geaflow-kubernetes-operator/geaflow-kubernetes-operator-web/web-dashboard/src/operatorAuth.ts
new file mode 100644
index 000000000..600dd6d25
--- /dev/null
+++ b/geaflow-kubernetes-operator/geaflow-kubernetes-operator-web/web-dashboard/src/operatorAuth.ts
@@ -0,0 +1,49 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+// Credentials live only in this tab's memory. Never persist them in browser storage or URLs.
+let token: string | undefined;
+const authChanged = 'geaflow-operator-auth-changed';
+
+export function getOperatorToken() {
+ return token;
+}
+
+export function setOperatorToken(value?: string) {
+ token = value;
+ window.dispatchEvent(new Event(authChanged));
+}
+
+export function onOperatorAuthChanged(listener: () => void) {
+ window.addEventListener(authChanged, listener);
+ return () => window.removeEventListener(authChanged, listener);
+}
+
+export function operatorApiUrl(path: string) {
+ if (!path.startsWith('/api/')) {
+ throw new Error('Invalid operator API path');
+ }
+ return new URL(`.${path}`, new URL('.', window.location.href)).href;
+}
+
+export function operatorAuthHeaders(url: string): Record {
+ const apiBase = operatorApiUrl('/api/');
+ const target = new URL(url, window.location.href).href;
+ return token && target.startsWith(apiBase) ? { Authorization: `Bearer ${token}` } : {};
+}
diff --git a/geaflow-kubernetes-operator/geaflow-kubernetes-operator-web/web-dashboard/src/requestErrorConfig.ts b/geaflow-kubernetes-operator/geaflow-kubernetes-operator-web/web-dashboard/src/requestErrorConfig.ts
index 1eef53d67..8addc1f79 100644
--- a/geaflow-kubernetes-operator/geaflow-kubernetes-operator-web/web-dashboard/src/requestErrorConfig.ts
+++ b/geaflow-kubernetes-operator/geaflow-kubernetes-operator-web/web-dashboard/src/requestErrorConfig.ts
@@ -1,4 +1,4 @@
-/*
+/*
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements. See the NOTICE file
* distributed with this work for additional information
@@ -20,14 +20,7 @@
import type { RequestOptions } from '@@/plugin-request/request';
import type { RequestConfig } from '@umijs/max';
import { message, notification } from 'antd';
-
-function getRealRequestUrl(url: string | undefined) {
- let pathname = window.location.pathname;
- if (pathname.endsWith("/")) {
- pathname = pathname.substring(0, pathname.length - 1);
- }
- return pathname + url;
-}
+import { operatorApiUrl, operatorAuthHeaders, setOperatorToken } from './operatorAuth';
// 错误处理方案： 错误类型
enum ErrorShowType {
@@ -99,6 +92,9 @@ export const errorConfig: RequestConfig = {
} else if (error.response) {
// Axios 的错误
// 请求成功发出且服务器也响应了状态码，但状态代码超出了 2xx 的范围
+ if (error.response.status === 401) {
+ setOperatorToken();
+ }
message.error(`Response status:${error.response.status}`);
} else if (error.request) {
// 请求已经成功发起，但没有收到响应
@@ -116,8 +112,8 @@ export const errorConfig: RequestConfig = {
requestInterceptors: [
(config: RequestOptions) => {
// 拦截请求配置，进行个性化处理。
- const url = getRealRequestUrl(config?.url);
- return { ...config, url };
+ const url = operatorApiUrl(config.url || '');
+ return { ...config, url, headers: { ...config.headers, ...operatorAuthHeaders(url) } };
},
],
diff --git a/geaflow-kubernetes-operator/helm/geaflow-kubernetes-operator/Chart.yaml b/geaflow-kubernetes-operator/helm/geaflow-kubernetes-operator/Chart.yaml
index b42714fae..c3d3e4231 100644
--- a/geaflow-kubernetes-operator/helm/geaflow-kubernetes-operator/Chart.yaml
+++ b/geaflow-kubernetes-operator/helm/geaflow-kubernetes-operator/Chart.yaml
@@ -34,7 +34,7 @@ type: application
# This is the chart version. This version number should be incremented each time you make changes
# to the chart and its templates, including the app version.
# Versions are expected to follow Semantic Versioning (https://semver.org/)
-version: 0.1.0
+version: 0.1.1
# This is the version number of the application being deployed. This version number should be
# incremented each time you make changes to the application. Versions are not expected to
diff --git a/geaflow-kubernetes-operator/helm/geaflow-kubernetes-operator/templates/deployment.yaml b/geaflow-kubernetes-operator/helm/geaflow-kubernetes-operator/templates/deployment.yaml
index d15ad4339..c29e1324a 100644
--- a/geaflow-kubernetes-operator/helm/geaflow-kubernetes-operator/templates/deployment.yaml
+++ b/geaflow-kubernetes-operator/helm/geaflow-kubernetes-operator/templates/deployment.yaml
@@ -24,6 +24,8 @@ metadata:
labels:
{{- include "geaflow-operator.labels" . | nindent 4 }}
spec:
+ strategy:
+ {{- toYaml .Values.strategy | nindent 4 }}
{{- if not .Values.autoscaling.enabled }}
replicas: {{ .Values.replicaCount }}
{{- end }}
@@ -50,7 +52,7 @@ spec:
- name: {{ .Chart.Name }}
securityContext:
{{- toYaml .Values.securityContext | nindent 12 }}
- image: "{{ .Values.image.repository }}:{{ .Values.image.tag | default .Chart.AppVersion }}"
+ image: "{{ .Values.image.repository }}:{{ required "image.tag must identify an image built with operator web authentication" .Values.image.tag }}"
imagePullPolicy: {{ .Values.image.pullPolicy }}
ports:
- name: http
@@ -73,10 +75,21 @@ spec:
resources:
{{- toYaml .Values.resources | nindent 12 }}
env:
- - name: USE_LOCAL_K8S
- value: "true"
+ - name: IS_LOCAL_MODE
+ value: "false"
- name: HTTP_SERVER_PORT
value: "{{ .Values.container.httpPort }}"
+ - name: GEAFLOW_OPERATOR_ADMIN_TOKEN
+ valueFrom:
+ secretKeyRef:
+ name: {{ required "webSecurity.existingSecret must reference a Secret containing the admin token" .Values.webSecurity.existingSecret | quote }}
+ key: {{ required "webSecurity.adminTokenKey must not be empty" .Values.webSecurity.adminTokenKey | quote }}
+ - name: GEAFLOW_OPERATOR_VIEWER_TOKEN
+ valueFrom:
+ secretKeyRef:
+ name: {{ .Values.webSecurity.existingSecret | quote }}
+ key: {{ required "webSecurity.viewerTokenKey must not be empty" .Values.webSecurity.viewerTokenKey | quote }}
+ optional: true
{{- with .Values.nodeSelector }}
nodeSelector:
{{- toYaml . | nindent 8 }}
diff --git a/geaflow-kubernetes-operator/helm/geaflow-kubernetes-operator/templates/networkpolicy.yaml b/geaflow-kubernetes-operator/helm/geaflow-kubernetes-operator/templates/networkpolicy.yaml
new file mode 100644
index 000000000..bd2d74e5e
--- /dev/null
+++ b/geaflow-kubernetes-operator/helm/geaflow-kubernetes-operator/templates/networkpolicy.yaml
@@ -0,0 +1,34 @@
+################################################################################
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+################################################################################
+
+{{- if .Values.networkPolicy.enabled }}
+apiVersion: networking.k8s.io/v1
+kind: NetworkPolicy
+metadata:
+ name: {{ include "geaflow-operator.fullname" . }}-web
+ labels:
+ {{- include "geaflow-operator.labels" . | nindent 4 }}
+spec:
+ podSelector:
+ matchLabels:
+ {{- include "geaflow-operator.selectorLabels" . | nindent 6 }}
+ policyTypes:
+ - Ingress
+ ingress: {{ toYaml .Values.networkPolicy.ingress | nindent 4 }}
+{{- end }}
diff --git a/geaflow-kubernetes-operator/helm/geaflow-kubernetes-operator/values.yaml b/geaflow-kubernetes-operator/helm/geaflow-kubernetes-operator/values.yaml
index c32145752..c847306ac 100644
--- a/geaflow-kubernetes-operator/helm/geaflow-kubernetes-operator/values.yaml
+++ b/geaflow-kubernetes-operator/helm/geaflow-kubernetes-operator/values.yaml
@@ -26,7 +26,7 @@ replicaCount: 1
image:
repository: geaflow-kubernetes-operator
pullPolicy: IfNotPresent
- # Overrides the image tag whose default is the chart appVersion.
+ # Required: use a unique tag for an image built with this operator's web authentication.
tag: ""
imagePullSecrets: [ ]
@@ -88,6 +88,19 @@ securityContext: { }
container:
httpPort: 8089
+# Reference a pre-created Secret in the release namespace. No credentials are stored in Helm values.
+webSecurity:
+ existingSecret: ""
+ adminTokenKey: admin-token
+ # Optional key: omit it from the Secret to disable read-only access.
+ viewerTokenKey: viewer-token
+
+# Requires a NetworkPolicy-capable CNI. Default: deny ingress; use kubectl port-forward.
+# Add explicit ingress rules for approved clients or a TLS ingress controller.
+networkPolicy:
+ enabled: true
+ ingress: []
+
service:
type: ClusterIP
port: 80
diff --git a/geaflow-kubernetes-operator/pom.xml b/geaflow-kubernetes-operator/pom.xml
index 2add05f46..1a15d66e4 100644
--- a/geaflow-kubernetes-operator/pom.xml
+++ b/geaflow-kubernetes-operator/pom.xml
@@ -43,6 +43,7 @@
3.2.22.0.2920.0
+ 0.8.0-SNAPSHOT111111
@@ -107,12 +108,6 @@
${junit.version}
-
- org.springframework.boot
- spring-boot-starter-test
- test
-
-
org.apache.geaflowgeaflow-kubernetes-operator-bootstrap
@@ -146,7 +141,7 @@
org.apache.geaflowgeaflow-on-k8s
- ${project.version}
+ ${geaflow.version}
@@ -157,12 +152,12 @@
org.apache.maven.pluginsmaven-checkstyle-plugin
- 2.17
+ 3.3.0checkstylecom.puppycrawl.tools
- 7.1
+ 8.45
@@ -212,11 +207,6 @@
org.apache.maven.surefire2.20
-
- surefire-testng
- org.apache.maven.surefire
- 2.20
-
@@ -226,9 +216,6 @@
**/*Tests.java**/*Test.java
-
- false
- 1
diff --git a/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-clickhouse/pom.xml b/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-clickhouse/pom.xml
index e5ec03cba..3ee498d53 100644
--- a/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-clickhouse/pom.xml
+++ b/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-clickhouse/pom.xml
@@ -34,7 +34,6 @@
0.6.55.2.1
- 1.19.8
@@ -72,4 +71,18 @@
test
+
+
+
+ org.apache.maven.plugins
+ maven-surefire-plugin
+
+
+
+ org.slf4j:slf4j-log4j12
+
+
+
+
+
diff --git a/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-clickhouse/src/test/java/org/apache/geaflow/dsl/connector/clickhouse/ClickHouseTableConnectorIntegrationTest.java b/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-clickhouse/src/test/java/org/apache/geaflow/dsl/connector/clickhouse/ClickHouseTableConnectorIntegrationTest.java
new file mode 100644
index 000000000..57c6ff4bd
--- /dev/null
+++ b/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-clickhouse/src/test/java/org/apache/geaflow/dsl/connector/clickhouse/ClickHouseTableConnectorIntegrationTest.java
@@ -0,0 +1,244 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.geaflow.dsl.connector.clickhouse;
+
+import java.io.IOException;
+import java.sql.Connection;
+import java.sql.DriverManager;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.time.Duration;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Locale;
+import java.util.Optional;
+import org.apache.geaflow.common.config.Configuration;
+import org.apache.geaflow.common.type.Types;
+import org.apache.geaflow.dsl.common.data.Row;
+import org.apache.geaflow.dsl.common.data.impl.ObjectRow;
+import org.apache.geaflow.dsl.common.types.StructType;
+import org.apache.geaflow.dsl.common.types.TableField;
+import org.apache.geaflow.dsl.common.types.TableSchema;
+import org.apache.geaflow.dsl.connector.api.FetchData;
+import org.apache.geaflow.dsl.connector.api.Partition;
+import org.apache.geaflow.dsl.connector.api.window.AllFetchWindow;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.testcontainers.DockerClientFactory;
+import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.containers.wait.strategy.Wait;
+import org.testcontainers.utility.DockerImageName;
+import org.testng.Assert;
+import org.testng.annotations.AfterClass;
+import org.testng.annotations.BeforeClass;
+import org.testng.annotations.Test;
+
+public class ClickHouseTableConnectorIntegrationTest {
+
+ private static final Logger LOGGER =
+ LoggerFactory.getLogger(ClickHouseTableConnectorIntegrationTest.class);
+
+ private static final DockerImageName IMAGE =
+ DockerImageName.parse("clickhouse/clickhouse-server:24.3");
+
+ private static final int ROWS = 500;
+
+ private static GenericContainer clickhouse;
+ private static String url;
+ private static String username;
+ private static String password;
+
+ @BeforeClass
+ public static void setup() throws SQLException {
+ boolean dockerRequired = ClickHouseTestEnvironment.isDockerRequired(
+ System.getProperty("clickhouse.it.required"), System.getenv("CI"));
+ ClickHouseTestEnvironment.requireDocker(dockerRequired,
+ () -> DockerClientFactory.instance().isDockerAvailable());
+ username = "ch_user";
+ password = "ch_pass";
+ clickhouse = new GenericContainer<>(IMAGE);
+ clickhouse.withExposedPorts(8123);
+ clickhouse.withEnv("CLICKHOUSE_USER", username);
+ clickhouse.withEnv("CLICKHOUSE_PASSWORD", password);
+ clickhouse.withEnv("CLICKHOUSE_DB", "default");
+ clickhouse.withEnv("CLICKHOUSE_DEFAULT_ACCESS_MANAGEMENT", "1");
+ clickhouse.waitingFor(Wait.forHttp("/ping").forStatusCode(200)
+ .withStartupTimeout(Duration.ofMinutes(2)));
+ clickhouse.start();
+ url = "jdbc:clickhouse://" + clickhouse.getHost() + ":"
+ + clickhouse.getMappedPort(8123) + "/default";
+ LOGGER.info("clickhouse started at {}", url);
+
+ try (Connection connection = DriverManager.getConnection(url, username, password);
+ Statement statement = connection.createStatement()) {
+ statement.execute("CREATE TABLE events (id Int64, name String, score Float64) "
+ + "ENGINE = MergeTree() ORDER BY id");
+ // Seed the read tests with a known, ordered data set.
+ StringBuilder insert = new StringBuilder("INSERT INTO events (id, name, score) VALUES ");
+ for (int i = 0; i < ROWS; i++) {
+ if (i > 0) {
+ insert.append(",");
+ }
+ insert.append(String.format(Locale.ROOT, "(%d, 'name-%d', %f)", i, i, (double) i));
+ }
+ statement.execute(insert.toString());
+ }
+ }
+
+ @AfterClass(alwaysRun = true)
+ public static void cleanup() {
+ if (clickhouse != null) {
+ clickhouse.stop();
+ }
+ }
+
+ private static TableField[] fields() {
+ return new TableField[]{
+ new TableField("id", Types.LONG, false),
+ new TableField("name", Types.BINARY_STRING, false),
+ new TableField("score", Types.DOUBLE, false)
+ };
+ }
+
+ private static Configuration baseConf(String tableName) {
+ Configuration conf = new Configuration();
+ conf.put(ClickHouseConfigKeys.GEAFLOW_DSL_CLICKHOUSE_URL, url);
+ conf.put(ClickHouseConfigKeys.GEAFLOW_DSL_CLICKHOUSE_USERNAME, username);
+ conf.put(ClickHouseConfigKeys.GEAFLOW_DSL_CLICKHOUSE_PASSWORD, password);
+ conf.put(ClickHouseConfigKeys.GEAFLOW_DSL_CLICKHOUSE_TABLE_NAME, tableName);
+ return conf;
+ }
+
+ private static long countRows(String tableName) throws SQLException {
+ try (Connection connection = DriverManager.getConnection(url, username, password);
+ Statement statement = connection.createStatement();
+ ResultSet rs = statement.executeQuery("SELECT count() FROM " + tableName)) {
+ rs.next();
+ return rs.getLong(1);
+ }
+ }
+
+ private static void createTable(String tableName, String engine) throws SQLException {
+ try (Connection connection = DriverManager.getConnection(url, username, password);
+ Statement statement = connection.createStatement()) {
+ statement.execute(String.format("CREATE TABLE %s (id Int64, name String, score Float64) "
+ + "ENGINE = %s", tableName, engine));
+ }
+ }
+
+ private static void writeRows(String tableName, int batchSize, int count) throws IOException {
+ Configuration conf = baseConf(tableName);
+ conf.put(ClickHouseConfigKeys.GEAFLOW_DSL_CLICKHOUSE_WRITE_BATCH_SIZE,
+ String.valueOf(batchSize));
+ ClickHouseTableSink sink = new ClickHouseTableSink();
+ sink.init(conf, new StructType(fields()));
+ sink.open(null);
+ try {
+ for (int i = 0; i < count; i++) {
+ sink.write(ObjectRow.create(new Object[]{(long) i, "name-" + i, (double) i}));
+ }
+ sink.finish();
+ } finally {
+ sink.close();
+ }
+ }
+
+ private static List readAll(Configuration conf) throws IOException {
+ ClickHouseTableSource source = new ClickHouseTableSource();
+ source.init(conf, new TableSchema(fields()));
+ source.open(null);
+ List rows = new ArrayList<>();
+ try {
+ for (Partition partition : source.listPartitions()) {
+ FetchData data = source.fetch(partition, Optional.empty(),
+ new AllFetchWindow(0L));
+ Iterator iterator = data.getDataIterator();
+ while (iterator.hasNext()) {
+ rows.add(iterator.next());
+ }
+ }
+ } finally {
+ source.close();
+ }
+ return rows;
+ }
+
+ @Test
+ public void testSinkBatchWrite() throws Exception {
+ createTable("sink_test", "MergeTree() ORDER BY id");
+ writeRows("sink_test", 100, ROWS);
+ Assert.assertEquals(countRows("sink_test"), ROWS);
+ }
+
+ @Test
+ public void testSourceReadAll() throws Exception {
+ Configuration conf = baseConf("events");
+ List rows = readAll(conf);
+ Assert.assertEquals(rows.size(), ROWS);
+ // Rows come back ordered by id; spot check that the full first row round-tripped.
+ Row first = rows.get(0);
+ Assert.assertEquals(((Number) first.getField(0, Types.LONG)).longValue(), 0L);
+ Assert.assertEquals(String.valueOf(first.getField(1, Types.BINARY_STRING)), "name-0");
+ Assert.assertEquals(((Number) first.getField(2, Types.DOUBLE)).doubleValue(), 0.0, 1e-9);
+ }
+
+ @Test
+ public void testSourceParallelPartitionedRead() throws Exception {
+ Configuration conf = baseConf("events");
+ conf.put(ClickHouseConfigKeys.GEAFLOW_DSL_CLICKHOUSE_PARTITION_NUM, "2");
+ conf.put(ClickHouseConfigKeys.GEAFLOW_DSL_CLICKHOUSE_PARTITION_COLUMN, "id");
+ conf.put(ClickHouseConfigKeys.GEAFLOW_DSL_CLICKHOUSE_PARTITION_LOWERBOUND, "0");
+ conf.put(ClickHouseConfigKeys.GEAFLOW_DSL_CLICKHOUSE_PARTITION_UPPERBOUND,
+ String.valueOf(ROWS));
+
+ ClickHouseTableSource source = new ClickHouseTableSource();
+ source.init(conf, new TableSchema(fields()));
+ source.open(null);
+ List partitions = source.listPartitions();
+ Assert.assertEquals(partitions.size(), 2);
+
+ int total = 0;
+ for (Partition partition : partitions) {
+ FetchData data = source.fetch(partition, Optional.empty(), new AllFetchWindow(0L));
+ Iterator iterator = data.getDataIterator();
+ while (iterator.hasNext()) {
+ iterator.next();
+ total++;
+ }
+ }
+ source.close();
+ // The two partitions cover disjoint id ranges; together they must return every row exactly once.
+ Assert.assertEquals(total, ROWS);
+ }
+
+ @Test
+ public void testBatchAndRowByRowPreserveRows() throws Exception {
+ createTable("batched_rows", "Memory");
+ createTable("row_by_row_rows", "Memory");
+ int rows = 65;
+ // Include a partial final batch; elapsed time on shared runners is not an assertion.
+ writeRows("batched_rows", 32, rows);
+ writeRows("row_by_row_rows", 1, rows);
+ Assert.assertEquals(countRows("batched_rows"), rows);
+ Assert.assertEquals(countRows("row_by_row_rows"), rows);
+ }
+}
diff --git a/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-clickhouse/src/test/java/org/apache/geaflow/dsl/connector/clickhouse/ClickHouseTableConnectorTest.java b/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-clickhouse/src/test/java/org/apache/geaflow/dsl/connector/clickhouse/ClickHouseTableConnectorTest.java
index b2c1530a3..f6906cd05 100644
--- a/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-clickhouse/src/test/java/org/apache/geaflow/dsl/connector/clickhouse/ClickHouseTableConnectorTest.java
+++ b/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-clickhouse/src/test/java/org/apache/geaflow/dsl/connector/clickhouse/ClickHouseTableConnectorTest.java
@@ -19,93 +19,19 @@
package org.apache.geaflow.dsl.connector.clickhouse;
-import java.io.IOException;
-import java.sql.Connection;
-import java.sql.DriverManager;
-import java.sql.ResultSet;
-import java.sql.SQLException;
-import java.sql.Statement;
-import java.util.ArrayList;
-import java.util.Iterator;
import java.util.List;
-import java.util.Locale;
-import java.util.Optional;
import org.apache.geaflow.common.config.Configuration;
import org.apache.geaflow.common.type.Types;
-import org.apache.geaflow.dsl.common.data.Row;
-import org.apache.geaflow.dsl.common.data.impl.ObjectRow;
-import org.apache.geaflow.dsl.common.types.StructType;
import org.apache.geaflow.dsl.common.types.TableField;
import org.apache.geaflow.dsl.common.types.TableSchema;
-import org.apache.geaflow.dsl.connector.api.FetchData;
import org.apache.geaflow.dsl.connector.api.Partition;
import org.apache.geaflow.dsl.connector.api.TableConnector;
import org.apache.geaflow.dsl.connector.api.util.ConnectorFactory;
-import org.apache.geaflow.dsl.connector.api.window.AllFetchWindow;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import org.testcontainers.containers.GenericContainer;
-import org.testcontainers.containers.wait.strategy.Wait;
-import org.testcontainers.utility.DockerImageName;
import org.testng.Assert;
-import org.testng.annotations.AfterClass;
-import org.testng.annotations.BeforeClass;
import org.testng.annotations.Test;
public class ClickHouseTableConnectorTest {
- private static final Logger LOGGER =
- LoggerFactory.getLogger(ClickHouseTableConnectorTest.class);
-
- private static final DockerImageName IMAGE =
- DockerImageName.parse("clickhouse/clickhouse-server:24.3");
-
- private static final int ROWS = 500;
-
- private static GenericContainer clickhouse;
- private static String url;
- private static String username;
- private static String password;
-
- @BeforeClass
- public static void setup() throws SQLException {
- username = "ch_user";
- password = "ch_pass";
- clickhouse = new GenericContainer<>(IMAGE);
- clickhouse.withExposedPorts(8123);
- clickhouse.withEnv("CLICKHOUSE_USER", username);
- clickhouse.withEnv("CLICKHOUSE_PASSWORD", password);
- clickhouse.withEnv("CLICKHOUSE_DB", "default");
- clickhouse.withEnv("CLICKHOUSE_DEFAULT_ACCESS_MANAGEMENT", "1");
- clickhouse.waitingFor(Wait.forHttp("/ping").forStatusCode(200));
- clickhouse.start();
- url = "jdbc:clickhouse://" + clickhouse.getHost() + ":"
- + clickhouse.getMappedPort(8123) + "/default";
- LOGGER.info("clickhouse started at {}", url);
-
- try (Connection connection = DriverManager.getConnection(url, username, password);
- Statement statement = connection.createStatement()) {
- statement.execute("CREATE TABLE events (id Int64, name String, score Float64) "
- + "ENGINE = MergeTree() ORDER BY id");
- // Seed the read tests with a known, ordered data set.
- StringBuilder insert = new StringBuilder("INSERT INTO events (id, name, score) VALUES ");
- for (int i = 0; i < ROWS; i++) {
- if (i > 0) {
- insert.append(",");
- }
- insert.append(String.format(Locale.ROOT, "(%d, 'name-%d', %f)", i, i, (double) i));
- }
- statement.execute(insert.toString());
- }
- }
-
- @AfterClass
- public static void cleanup() {
- if (clickhouse != null) {
- clickhouse.stop();
- }
- }
-
private static TableField[] fields() {
return new TableField[]{
new TableField("id", Types.LONG, false),
@@ -116,115 +42,11 @@ private static TableField[] fields() {
private static Configuration baseConf(String tableName) {
Configuration conf = new Configuration();
- conf.put(ClickHouseConfigKeys.GEAFLOW_DSL_CLICKHOUSE_URL, url);
- conf.put(ClickHouseConfigKeys.GEAFLOW_DSL_CLICKHOUSE_USERNAME, username);
- conf.put(ClickHouseConfigKeys.GEAFLOW_DSL_CLICKHOUSE_PASSWORD, password);
+ conf.put(ClickHouseConfigKeys.GEAFLOW_DSL_CLICKHOUSE_URL, "jdbc:clickhouse://localhost:8123/default");
conf.put(ClickHouseConfigKeys.GEAFLOW_DSL_CLICKHOUSE_TABLE_NAME, tableName);
return conf;
}
- private static long countRows(String tableName) throws SQLException {
- try (Connection connection = DriverManager.getConnection(url, username, password);
- Statement statement = connection.createStatement();
- ResultSet rs = statement.executeQuery("SELECT count() FROM " + tableName)) {
- rs.next();
- return rs.getLong(1);
- }
- }
-
- private static void createTable(String tableName, String engine) throws SQLException {
- try (Connection connection = DriverManager.getConnection(url, username, password);
- Statement statement = connection.createStatement()) {
- statement.execute(String.format("CREATE TABLE %s (id Int64, name String, score Float64) "
- + "ENGINE = %s", tableName, engine));
- }
- }
-
- private static void writeRows(String tableName, int batchSize, int count) throws IOException {
- Configuration conf = baseConf(tableName);
- conf.put(ClickHouseConfigKeys.GEAFLOW_DSL_CLICKHOUSE_WRITE_BATCH_SIZE,
- String.valueOf(batchSize));
- ClickHouseTableSink sink = new ClickHouseTableSink();
- sink.init(conf, new StructType(fields()));
- sink.open(null);
- try {
- for (int i = 0; i < count; i++) {
- sink.write(ObjectRow.create(new Object[]{(long) i, "name-" + i, (double) i}));
- }
- sink.finish();
- } finally {
- sink.close();
- }
- }
-
- private static List readAll(Configuration conf) throws IOException {
- ClickHouseTableSource source = new ClickHouseTableSource();
- source.init(conf, new TableSchema(fields()));
- source.open(null);
- List rows = new ArrayList<>();
- try {
- for (Partition partition : source.listPartitions()) {
- FetchData data = source.fetch(partition, Optional.empty(),
- new AllFetchWindow(0L));
- Iterator iterator = data.getDataIterator();
- while (iterator.hasNext()) {
- rows.add(iterator.next());
- }
- }
- } finally {
- source.close();
- }
- return rows;
- }
-
- @Test
- public void testSinkBatchWrite() throws Exception {
- createTable("sink_test", "MergeTree() ORDER BY id");
- writeRows("sink_test", 100, ROWS);
- Assert.assertEquals(countRows("sink_test"), ROWS);
- }
-
- @Test
- public void testSourceReadAll() throws Exception {
- Configuration conf = baseConf("events");
- List rows = readAll(conf);
- Assert.assertEquals(rows.size(), ROWS);
- // Rows come back ordered by id; spot check that the full first row round-tripped.
- Row first = rows.get(0);
- Assert.assertEquals(((Number) first.getField(0, Types.LONG)).longValue(), 0L);
- Assert.assertEquals(String.valueOf(first.getField(1, Types.BINARY_STRING)), "name-0");
- Assert.assertEquals(((Number) first.getField(2, Types.DOUBLE)).doubleValue(), 0.0, 1e-9);
- }
-
- @Test
- public void testSourceParallelPartitionedRead() throws Exception {
- Configuration conf = baseConf("events");
- conf.put(ClickHouseConfigKeys.GEAFLOW_DSL_CLICKHOUSE_PARTITION_NUM, "2");
- conf.put(ClickHouseConfigKeys.GEAFLOW_DSL_CLICKHOUSE_PARTITION_COLUMN, "id");
- conf.put(ClickHouseConfigKeys.GEAFLOW_DSL_CLICKHOUSE_PARTITION_LOWERBOUND, "0");
- conf.put(ClickHouseConfigKeys.GEAFLOW_DSL_CLICKHOUSE_PARTITION_UPPERBOUND,
- String.valueOf(ROWS));
-
- ClickHouseTableSource source = new ClickHouseTableSource();
- source.init(conf, new TableSchema(fields()));
- source.open(null);
- List partitions = source.listPartitions();
- Assert.assertEquals(partitions.size(), 2);
-
- int total = 0;
- for (Partition partition : partitions) {
- FetchData data = source.fetch(partition, Optional.empty(), new AllFetchWindow(0L));
- Iterator iterator = data.getDataIterator();
- while (iterator.hasNext()) {
- iterator.next();
- total++;
- }
- }
- source.close();
- // The two partitions cover disjoint id ranges; together they must return every row exactly once.
- Assert.assertEquals(total, ROWS);
- }
-
@Test
public void testListPartitionsSplitsRange() {
Configuration conf = baseConf("events");
@@ -248,33 +70,4 @@ public void testConnectorDiscoveredViaSpi() {
Assert.assertTrue(connector instanceof ClickHouseTableConnector);
Assert.assertEquals(connector.getType(), ClickHouseTableConnector.TYPE);
}
-
- @Test
- public void testBatchWriteOutperformsRowByRow() throws Exception {
- createTable("bench_batched", "Memory");
- createTable("bench_rowbyrow", "Memory");
- int benchRows = 2000;
-
- long batchedStart = System.nanoTime();
- writeRows("bench_batched", 2000, benchRows);
- long batchedNanos = System.nanoTime() - batchedStart;
-
- // batch size 1 flushes every row, i.e. one insert round-trip per row (the row-by-row baseline).
- long rowByRowStart = System.nanoTime();
- writeRows("bench_rowbyrow", 1, benchRows);
- long rowByRowNanos = System.nanoTime() - rowByRowStart;
-
- Assert.assertEquals(countRows("bench_batched"), benchRows);
- Assert.assertEquals(countRows("bench_rowbyrow"), benchRows);
-
- double batchedPerSec = benchRows / (batchedNanos / 1_000_000_000.0);
- double rowByRowPerSec = benchRows / (rowByRowNanos / 1_000_000_000.0);
- LOGGER.info("clickhouse write benchmark for {} rows: batched={} rows/s, "
- + "row-by-row={} rows/s, speedup={}x",
- benchRows, String.format("%.0f", batchedPerSec), String.format("%.0f", rowByRowPerSec),
- String.format("%.1f", batchedPerSec / rowByRowPerSec));
-
- Assert.assertTrue(batchedNanos < rowByRowNanos,
- "batched write should be faster than row-by-row");
- }
}
diff --git a/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-clickhouse/src/test/java/org/apache/geaflow/dsl/connector/clickhouse/ClickHouseTableSinkTest.java b/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-clickhouse/src/test/java/org/apache/geaflow/dsl/connector/clickhouse/ClickHouseTableSinkTest.java
new file mode 100644
index 000000000..b4e66dc8d
--- /dev/null
+++ b/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-clickhouse/src/test/java/org/apache/geaflow/dsl/connector/clickhouse/ClickHouseTableSinkTest.java
@@ -0,0 +1,115 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.geaflow.dsl.connector.clickhouse;
+
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+
+import java.lang.reflect.Field;
+import java.sql.Connection;
+import java.sql.PreparedStatement;
+import java.sql.SQLException;
+import org.apache.geaflow.common.config.Configuration;
+import org.apache.geaflow.common.type.Types;
+import org.apache.geaflow.dsl.common.data.impl.ObjectRow;
+import org.apache.geaflow.dsl.common.exception.GeaFlowDSLException;
+import org.apache.geaflow.dsl.common.types.StructType;
+import org.apache.geaflow.dsl.common.types.TableField;
+import org.testng.Assert;
+import org.testng.annotations.DataProvider;
+import org.testng.annotations.Test;
+
+public class ClickHouseTableSinkTest {
+
+ @DataProvider(name = "batchSizes")
+ public Object[][] batchSizes() {
+ return new Object[][]{{1, 7, 7}, {3, 7, 3}, {100, 7, 1}, {3, 0, 0}, {3, 6, 2}};
+ }
+
+ @Test(dataProvider = "batchSizes")
+ public void batchesRowsIntoExpectedNumberOfDatabaseCalls(int batchSize, int rows, int batches) throws Exception {
+ PreparedStatement statement = mock(PreparedStatement.class);
+ ClickHouseTableSink sink = sink(batchSize, statement, mock(Connection.class));
+ for (int i = 0; i < rows; i++) {
+ sink.write(ObjectRow.create(new Object[]{(long) i}));
+ }
+ sink.finish();
+ sink.finish();
+ verify(statement, times(rows)).addBatch();
+ verify(statement, times(batches)).executeBatch();
+ verify(statement, times(batches)).clearBatch();
+ sink.close();
+ }
+
+ @Test
+ public void failedBatchIsNotSilentlyResubmittedByFinish() throws Exception {
+ PreparedStatement statement = mock(PreparedStatement.class);
+ doThrow(new SQLException("simulated insert failure")).when(statement).executeBatch();
+ ClickHouseTableSink sink = sink(1, statement, mock(Connection.class));
+ try {
+ sink.write(ObjectRow.create(new Object[]{1L}));
+ Assert.fail("An insert failure must be reported");
+ } catch (GeaFlowDSLException expected) {
+ Assert.assertTrue(expected.getCause() instanceof SQLException);
+ }
+ sink.finish();
+ verify(statement, times(1)).executeBatch();
+ verify(statement, times(1)).clearBatch();
+ sink.close();
+ }
+
+ @Test
+ public void statementCloseFailureStillClosesConnection() throws Exception {
+ PreparedStatement statement = mock(PreparedStatement.class);
+ Connection connection = mock(Connection.class);
+ doThrow(new SQLException("simulated close failure")).when(statement).close();
+ ClickHouseTableSink sink = sink(1, statement, connection);
+ try {
+ sink.close();
+ Assert.fail("A close failure must be reported");
+ } catch (GeaFlowDSLException expected) {
+ Assert.assertTrue(expected.getCause() instanceof SQLException);
+ }
+ sink.close();
+ verify(connection, times(1)).close();
+ verify(statement, times(1)).close();
+ }
+
+ private static ClickHouseTableSink sink(int batchSize, PreparedStatement statement,
+ Connection connection) throws Exception {
+ Configuration conf = new Configuration();
+ conf.put(ClickHouseConfigKeys.GEAFLOW_DSL_CLICKHOUSE_URL, "jdbc:clickhouse://localhost:8123/default");
+ conf.put(ClickHouseConfigKeys.GEAFLOW_DSL_CLICKHOUSE_TABLE_NAME, "events");
+ conf.put(ClickHouseConfigKeys.GEAFLOW_DSL_CLICKHOUSE_WRITE_BATCH_SIZE, String.valueOf(batchSize));
+ ClickHouseTableSink sink = new ClickHouseTableSink();
+ sink.init(conf, new StructType(new TableField("id", Types.LONG, false)));
+ setField(sink, "statement", statement);
+ setField(sink, "connection", connection);
+ return sink;
+ }
+
+ private static void setField(ClickHouseTableSink sink, String name, Object value) throws Exception {
+ Field field = ClickHouseTableSink.class.getDeclaredField(name);
+ field.setAccessible(true);
+ field.set(sink, value);
+ }
+}
diff --git a/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-clickhouse/src/test/java/org/apache/geaflow/dsl/connector/clickhouse/ClickHouseTestEnvironment.java b/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-clickhouse/src/test/java/org/apache/geaflow/dsl/connector/clickhouse/ClickHouseTestEnvironment.java
new file mode 100644
index 000000000..2f462253c
--- /dev/null
+++ b/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-clickhouse/src/test/java/org/apache/geaflow/dsl/connector/clickhouse/ClickHouseTestEnvironment.java
@@ -0,0 +1,64 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.geaflow.dsl.connector.clickhouse;
+
+import java.util.function.BooleanSupplier;
+import org.testng.SkipException;
+
+final class ClickHouseTestEnvironment {
+
+ private static final String MESSAGE = "ClickHouse integration tests need a reachable Docker daemon. "
+ + "Check docker version, docker info, socket permissions and DOCKER_HOST. "
+ + "CI requires these tests; use a supported Docker runtime and Testcontainers client.";
+
+ private ClickHouseTestEnvironment() {
+ }
+
+ static boolean isDockerRequired(String configured, String ci) {
+ if ("true".equalsIgnoreCase(configured)) {
+ return true;
+ }
+ if ("false".equalsIgnoreCase(configured)) {
+ return false;
+ }
+ if (configured != null) {
+ throw new IllegalArgumentException("clickhouse.it.required must be true or false");
+ }
+ return ci != null && !ci.isEmpty() && !"false".equalsIgnoreCase(ci) && !"0".equals(ci);
+ }
+
+ static void requireDocker(boolean required, BooleanSupplier dockerAvailable) {
+ boolean available;
+ try {
+ available = dockerAvailable.getAsBoolean();
+ } catch (RuntimeException error) {
+ if (required) {
+ throw new IllegalStateException(MESSAGE, error);
+ }
+ throw new SkipException(MESSAGE, error);
+ }
+ if (!available) {
+ if (required) {
+ throw new IllegalStateException(MESSAGE);
+ }
+ throw new SkipException(MESSAGE + " Skipping optional local integration tests.");
+ }
+ }
+}
diff --git a/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-clickhouse/src/test/java/org/apache/geaflow/dsl/connector/clickhouse/ClickHouseTestEnvironmentTest.java b/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-clickhouse/src/test/java/org/apache/geaflow/dsl/connector/clickhouse/ClickHouseTestEnvironmentTest.java
new file mode 100644
index 000000000..16e4e1f2d
--- /dev/null
+++ b/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-clickhouse/src/test/java/org/apache/geaflow/dsl/connector/clickhouse/ClickHouseTestEnvironmentTest.java
@@ -0,0 +1,86 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.apache.geaflow.dsl.connector.clickhouse;
+
+import org.testng.Assert;
+import org.testng.SkipException;
+import org.testng.annotations.Test;
+
+public class ClickHouseTestEnvironmentTest {
+
+ @Test
+ public void resolvesLocalAndCiDefaultsAndExplicitOverrides() {
+ Assert.assertFalse(ClickHouseTestEnvironment.isDockerRequired(null, null));
+ Assert.assertFalse(ClickHouseTestEnvironment.isDockerRequired(null, ""));
+ Assert.assertFalse(ClickHouseTestEnvironment.isDockerRequired(null, "false"));
+ Assert.assertFalse(ClickHouseTestEnvironment.isDockerRequired(null, "0"));
+ Assert.assertTrue(ClickHouseTestEnvironment.isDockerRequired(null, "true"));
+ Assert.assertTrue(ClickHouseTestEnvironment.isDockerRequired(null, "1"));
+ Assert.assertTrue(ClickHouseTestEnvironment.isDockerRequired("true", null));
+ Assert.assertFalse(ClickHouseTestEnvironment.isDockerRequired("false", "true"));
+ }
+
+ @Test(expectedExceptions = IllegalArgumentException.class)
+ public void malformedRequiredFlagCannotSilentlyDisableTests() {
+ ClickHouseTestEnvironment.isDockerRequired("tru", "true");
+ }
+
+ @Test
+ public void availableDockerRunsInBothModes() {
+ ClickHouseTestEnvironment.requireDocker(false, () -> true);
+ ClickHouseTestEnvironment.requireDocker(true, () -> true);
+ }
+
+ @Test(expectedExceptions = SkipException.class)
+ public void unavailableDockerSkipsOnlyOptionalLocalTests() {
+ ClickHouseTestEnvironment.requireDocker(false, () -> false);
+ }
+
+ @Test(expectedExceptions = IllegalStateException.class)
+ public void unavailableRequiredDockerFails() {
+ ClickHouseTestEnvironment.requireDocker(true, () -> false);
+ }
+
+ @Test
+ public void discoveryErrorsFailRequiredTestsWithOriginalCause() {
+ RuntimeException failure = new IllegalStateException("Docker client API incompatible");
+ try {
+ ClickHouseTestEnvironment.requireDocker(true, () -> {
+ throw failure;
+ });
+ Assert.fail("A required Docker discovery error must fail the test");
+ } catch (IllegalStateException error) {
+ Assert.assertSame(error.getCause(), failure);
+ }
+ }
+
+ @Test
+ public void discoveryErrorsSkipOptionalTestsWithOriginalCause() {
+ RuntimeException failure = new IllegalStateException("Docker socket unavailable");
+ try {
+ ClickHouseTestEnvironment.requireDocker(false, () -> {
+ throw failure;
+ });
+ Assert.fail("An optional test without Docker must be marked skipped");
+ } catch (SkipException error) {
+ Assert.assertSame(error.getCause(), failure);
+ }
+ }
+}
diff --git a/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-clickhouse/src/test/resources/log4j2-test.xml b/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-clickhouse/src/test/resources/log4j2-test.xml
new file mode 100644
index 000000000..5ebd0eeb7
--- /dev/null
+++ b/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-clickhouse/src/test/resources/log4j2-test.xml
@@ -0,0 +1,33 @@
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
+
diff --git a/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/pom.xml b/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/pom.xml
index ef389691d..66343661d 100644
--- a/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/pom.xml
+++ b/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-doris/pom.xml
@@ -33,7 +33,6 @@
4.5.13
- 1.19.7
diff --git a/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-mongodb/pom.xml b/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-mongodb/pom.xml
index 7ea2b12a2..a954c07f8 100644
--- a/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-mongodb/pom.xml
+++ b/geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-mongodb/pom.xml
@@ -33,7 +33,6 @@
4.11.1
- 1.19.8
diff --git a/geaflow/geaflow-dsl/geaflow-dsl-connector/pom.xml b/geaflow/geaflow-dsl/geaflow-dsl-connector/pom.xml
index 2feb05b35..bfa99e996 100644
--- a/geaflow/geaflow-dsl/geaflow-dsl-connector/pom.xml
+++ b/geaflow/geaflow-dsl/geaflow-dsl-connector/pom.xml
@@ -54,6 +54,11 @@
geaflow-dsl-connector-clickhouse
+
+
+ 1.21.4
+
+
diff --git a/tools/ci/check-docker.sh b/tools/ci/check-docker.sh
new file mode 100644
index 000000000..aa9ca19d3
--- /dev/null
+++ b/tools/ci/check-docker.sh
@@ -0,0 +1,36 @@
+#!/usr/bin/env bash
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+#
+
+set -euo pipefail
+
+if ! command -v docker >/dev/null 2>&1; then
+ echo "Docker CLI is required for the container integration tests." >&2
+ exit 1
+fi
+
+# A client binary alone does not prove that the daemon is reachable.
+if ! docker version; then
+ echo "Docker daemon is unreachable. Check the service, socket permissions and DOCKER_HOST." >&2
+ exit 1
+fi
+if ! docker info; then
+ echo "Docker daemon health check failed; container integration tests cannot run." >&2
+ exit 1
+fi
diff --git a/tools/ci/test_check_docker.py b/tools/ci/test_check_docker.py
new file mode 100644
index 000000000..9c9197493
--- /dev/null
+++ b/tools/ci/test_check_docker.py
@@ -0,0 +1,75 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+#
+
+import os
+from pathlib import Path
+import subprocess
+import tempfile
+import unittest
+
+
+class DockerPreflightTest(unittest.TestCase):
+ def run_preflight(self, installed=True, fail_command=""):
+ with tempfile.TemporaryDirectory() as directory:
+ directory = Path(directory)
+ calls = directory / "calls"
+ if installed:
+ docker = directory / "docker"
+ docker.write_text(
+ "#!/bin/bash\n"
+ 'printf "%s\\n" "$*" >> "$DOCKER_TEST_CALLS"\n'
+ 'if [[ "$1" == "$DOCKER_TEST_FAIL" ]]; then exit 1; fi\n'
+ 'echo "Docker $1 succeeded"\n'
+ )
+ docker.chmod(0o755)
+ environment = dict(os.environ)
+ environment.update({"PATH": str(directory), "DOCKER_TEST_CALLS": str(calls),
+ "DOCKER_TEST_FAIL": fail_command})
+ result = subprocess.run(
+ ["/bin/bash", str(Path(__file__).with_name("check-docker.sh"))],
+ env=environment, capture_output=True, text=True,
+ )
+ return result, calls.read_text().splitlines() if calls.exists() else []
+
+ def test_missing_cli_fails(self):
+ result, calls = self.run_preflight(installed=False)
+ self.assertNotEqual(result.returncode, 0)
+ self.assertIn("Docker CLI is required", result.stderr)
+ self.assertEqual(calls, [])
+
+ def test_unreachable_daemon_fails_before_info(self):
+ result, calls = self.run_preflight(fail_command="version")
+ self.assertNotEqual(result.returncode, 0)
+ self.assertIn("Docker daemon is unreachable", result.stderr)
+ self.assertEqual(calls, ["version"])
+
+ def test_unhealthy_daemon_fails(self):
+ result, calls = self.run_preflight(fail_command="info")
+ self.assertNotEqual(result.returncode, 0)
+ self.assertIn("Docker daemon health check failed", result.stderr)
+ self.assertEqual(calls, ["version", "info"])
+
+ def test_healthy_daemon_passes(self):
+ result, calls = self.run_preflight()
+ self.assertEqual(result.returncode, 0, result.stderr)
+ self.assertEqual(calls, ["version", "info"])
+
+
+if __name__ == "__main__":
+ unittest.main()