From f336aa76507b620fa5ac6bfb31ff2af7d4a58b0d Mon Sep 17 00:00:00 2001 From: kitalkuyo-gita Date: Fri, 30 Oct 2026 19:59:09 +0800 Subject: [PATCH 1/6] fix(operator): configure web access and deployment defaults --- SECURITY.md | 6 +- geaflow-kubernetes-operator/README.md | 24 +- geaflow-kubernetes-operator/README_cn.md | 24 +- geaflow-kubernetes-operator/SECURITY_MODEL.md | 168 +++++++++++++ .../src/main/resources/application.properties | 7 +- .../geaflow-kubernetes-operator-web/pom.xml | 29 +++ .../web/controller/api/JobController.java | 7 + .../web/security/BearerTokenFilter.java | 63 +++++ .../operator/web/security/OperatorTokens.java | 86 +++++++ .../web/security/SecurityResponse.java | 53 +++++ .../security/WebSecurityConfiguration.java | 87 +++++++ .../web/security/OperatorTokensTest.java | 76 ++++++ .../web/security/WebSecurityTest.java | 220 ++++++++++++++++++ .../web-dashboard/src/app.tsx | 5 +- .../src/components/OperatorAccess/index.tsx | 77 ++++++ .../web-dashboard/src/operatorAuth.test.ts | 59 +++++ .../web-dashboard/src/operatorAuth.ts | 49 ++++ .../web-dashboard/src/requestErrorConfig.ts | 18 +- .../geaflow-kubernetes-operator/Chart.yaml | 2 +- .../templates/NOTES.txt | 32 +-- .../templates/deployment.yaml | 15 +- .../templates/networkpolicy.yaml | 34 +++ .../geaflow-kubernetes-operator/values.yaml | 15 +- 23 files changed, 1106 insertions(+), 50 deletions(-) create mode 100644 geaflow-kubernetes-operator/SECURITY_MODEL.md create mode 100644 geaflow-kubernetes-operator/geaflow-kubernetes-operator-web/src/main/java/org/apache/geaflow/kubernetes/operator/web/security/BearerTokenFilter.java create mode 100644 geaflow-kubernetes-operator/geaflow-kubernetes-operator-web/src/main/java/org/apache/geaflow/kubernetes/operator/web/security/OperatorTokens.java create mode 100644 geaflow-kubernetes-operator/geaflow-kubernetes-operator-web/src/main/java/org/apache/geaflow/kubernetes/operator/web/security/SecurityResponse.java create mode 100644 geaflow-kubernetes-operator/geaflow-kubernetes-operator-web/src/main/java/org/apache/geaflow/kubernetes/operator/web/security/WebSecurityConfiguration.java create mode 100644 geaflow-kubernetes-operator/geaflow-kubernetes-operator-web/src/test/java/org/apache/geaflow/kubernetes/operator/web/security/OperatorTokensTest.java create mode 100644 geaflow-kubernetes-operator/geaflow-kubernetes-operator-web/src/test/java/org/apache/geaflow/kubernetes/operator/web/security/WebSecurityTest.java create mode 100644 geaflow-kubernetes-operator/geaflow-kubernetes-operator-web/web-dashboard/src/components/OperatorAccess/index.tsx create mode 100644 geaflow-kubernetes-operator/geaflow-kubernetes-operator-web/web-dashboard/src/operatorAuth.test.ts create mode 100644 geaflow-kubernetes-operator/geaflow-kubernetes-operator-web/web-dashboard/src/operatorAuth.ts create mode 100644 geaflow-kubernetes-operator/helm/geaflow-kubernetes-operator/templates/networkpolicy.yaml diff --git a/SECURITY.md b/SECURITY.md index 02ddfd373..3c638d9b9 100644 --- a/SECURITY.md +++ b/SECURITY.md @@ -10,7 +10,7 @@ If you discover a security vulnerability in Apache GeaFlow (Incubating), please Please send your security vulnerability report to the Apache Security Team at: -**[dev@geaflow.apache.org](mailto:dev@geaflow.apache.org)** +**[security@apache.org](mailto:security@apache.org)** You can also report to the GeaFlow project team directly at: @@ -55,3 +55,7 @@ Security fixes will be released as part of regular GeaFlow releases and announce - [Apache Security Team](https://www.apache.org/security/) - [Apache GeaFlow Official Website](https://geaflow.apache.org/) - [Apache GeaFlow GitHub Repository](https://github.com/apache/geaflow) + +## Security Model + +- [Kubernetes operator security model and deployment requirements](geaflow-kubernetes-operator/SECURITY_MODEL.md) diff --git a/geaflow-kubernetes-operator/README.md b/geaflow-kubernetes-operator/README.md index 56646d1fa..7bf0e52f5 100644 --- a/geaflow-kubernetes-operator/README.md +++ b/geaflow-kubernetes-operator/README.md @@ -13,14 +13,28 @@ cd geaflow-kubernetes-operator ./build-operator.sh ``` 2. The image name and tag can be modified in helm/geaflow-kubernetes-operator/values. -3. Execute this command in the project root directory to quickly deploy to the kubernetes cluster. +3. Create the required authentication Secret following [the security model](SECURITY_MODEL.md). + Set a unique tag for an image built from this source. Then deploy: ```shell cd geaflow-kubernetes-operator -helm install geaflow-kubernetes-operator helm/geaflow-kubernetes-operator +helm install geaflow-kubernetes-operator helm/geaflow-kubernetes-operator \ + --namespace geaflow-system \ + --set webSecurity.existingSecret=geaflow-operator-web-auth \ + --set image.tag=YOUR_UNIQUE_PATCHED_IMAGE_TAG \ + --set-json 'watchNamespaces=["geaflow-system"]' ``` ### Frontend Dashboard -Geaflow-kubernetes-operator comes with a frontend dashboard page, which you can access through port 8089. -The dashboard will display job statistics within the cluster, as well as their current status and basic information. +The dashboard requires an administrator or read-only bearer token. Prefer the read-only token. +The web service listens on port 8089 in Helm (8080 for standalone execution). The default +NetworkPolicy denies pod ingress; use a local tunnel and open `http://127.0.0.1:8089/`: + +```sh +kubectl -n geaflow-system port-forward --address 127.0.0.1 service/geaflow-kubernetes-operator 8089:80 +``` + +The dashboard displays job statistics and state. Tokens stay in page memory and must be +re-entered after reload. For TLS, credential rotation, RBAC and admission requirements, +read [the security model](SECURITY_MODEL.md). Job submission is reserved for trusted operators. ### Example @@ -29,5 +43,5 @@ Please refer to example/example.yml and modify the necessary parameters. After h is completed, execute this command to submit the example job into your kubernetes cluster. ```shell cd geaflow-kubernetes-operator -kubectl apply -f example/example_hla.yml +kubectl -n geaflow-system apply -f example/example_hla.yml ``` diff --git a/geaflow-kubernetes-operator/README_cn.md b/geaflow-kubernetes-operator/README_cn.md index 28f259180..2ddee031d 100644 --- a/geaflow-kubernetes-operator/README_cn.md +++ b/geaflow-kubernetes-operator/README_cn.md @@ -12,20 +12,34 @@ cd geaflow-kubernetes-operator ./build-operator.sh ``` 2. 在helm/geaflow-kubernetes-operator/values中可以修改镜像名和tag。 -3. 在项目根目录下执行以下命令即可快速部署到kubernetes集群中。 +3. 先按[安全模型与部署要求](SECURITY_MODEL.md)在部署命名空间创建认证 Secret， + 并使用包含本次认证功能的独立镜像 tag，再执行部署： ```shell cd geaflow-kubernetes-operator -helm install geaflow-kubernetes-operator helm/geaflow-kubernetes-operator +helm install geaflow-kubernetes-operator helm/geaflow-kubernetes-operator \ + --namespace geaflow-system \ + --set webSecurity.existingSecret=geaflow-operator-web-auth \ + --set image.tag=YOUR_UNIQUE_PATCHED_IMAGE_TAG \ + --set-json 'watchNamespaces=["geaflow-system"]' ``` ### 前端Dashboard -Geaflow-kubernetes-operator自带一个前端dashboard页面，你可以通过8089端口访问它。 -dashboard会展示集群内的作业统计，并展示它们的当前状态和基础信息。 +Dashboard 需要管理员或只读 Bearer 令牌，日常查看优先使用只读令牌。 +Helm 中 Web 监听端口为 8089（独立启动默认 8080）。默认 NetworkPolicy 拒绝 Pod 入站访问， +可以建立本机隧道并访问 `http://127.0.0.1:8089/`： + +```sh +kubectl -n geaflow-system port-forward --address 127.0.0.1 service/geaflow-kubernetes-operator 8089:80 +``` + +Dashboard 展示作业统计、状态和基础信息。令牌只保存在页面内存，刷新后需重新输入。 +管理员提交作业等同于授权执行代码，只能交给可信运维人员。 +TLS、Secret 轮换、命名空间 RBAC 和准入策略要求见[安全模型](SECURITY_MODEL.md)。 ### 示例 请参考example/example.yml并修改必要参数。在helm部署完成后，执行以下命令提交示例的作业。 ```shell cd geaflow-kubernetes-operator -kubectl apply -f example/example_hla.yml +kubectl -n geaflow-system apply -f example/example_hla.yml ``` diff --git a/geaflow-kubernetes-operator/SECURITY_MODEL.md b/geaflow-kubernetes-operator/SECURITY_MODEL.md new file mode 100644 index 000000000..f79f68010 --- /dev/null +++ b/geaflow-kubernetes-operator/SECURITY_MODEL.md @@ -0,0 +1,168 @@ +# Kubernetes operator security model + +This model describes the operator with mandatory web authentication. It does not apply +retroactively to older operator images. Installing this Helm chart around an old image +cannot add application authentication. + +## Trust and permissions + +The operator is a privileged workload deployment service. Only trusted cluster operators +may hold its administrator token or create/update GeaFlow custom resources directly. +Submitting a job means authorizing execution of its image, entry class and executable +artifacts, with the job's service account. Job configuration can influence deployment +settings as well. An administrator token is not a safe capability to give to untrusted +code authors or isolated tenants. + +The administrator token permits all web API methods. The optional, separate viewer token +permits only GET and HEAD under `/api`. Viewers can see all returned job specifications +and cluster information in the operator's namespace; those responses may contain sensitive +configuration. GET/HEAD endpoints must remain free of mutations. There is no per-job, +per-tenant or individual-user authorization. Tokens are shared deployment credentials, +not Kubernetes credentials, and requests are not impersonated into Kubernetes. + +Unauthenticated network clients, including other pods, must not read or mutate business +API resources. Only the GET health response and dashboard static files are public. +Missing, malformed, too-short or identical administrator/viewer credentials prevent startup. +Invalid/missing request credentials return HTTP 401; an authenticated viewer attempting a +write receives HTTP 403. No anonymous fallback or authentication-disable option is provided. + +## Configure credentials + +Generate independent random tokens with at least 32 random bytes (for example, +`openssl rand -hex 32`). The accepted representation is 32–512 characters using RFC 6750 +bearer-token characters. Length validation cannot establish entropy; do not use repeated +characters, example strings or passwords. + +For standalone execution set `GEAFLOW_OPERATOR_ADMIN_TOKEN` and optionally +`GEAFLOW_OPERATOR_VIEWER_TOKEN` in the process environment. The equivalent Spring properties +are `geaflow.operator.web.security.admin-token` and `geaflow.operator.web.security.viewer-token`. +Use a secret manager, protected environment or Spring configuration tree. Do not put +credentials in Git, Helm values, command-line arguments, URLs or application logs. + +For Kubernetes, create a Secret in the operator's namespace. This example keeps token values +out of command arguments and terminal output. Protect and distribute the files through your +approved secret-management process; the administrator and viewer tokens have different trust +requirements. Disable shell tracing before handling real tokens. + +```sh +set +x +umask 077 +OPERATOR_SECRET_DIR=$(mktemp -d) +openssl rand -hex 32 | tr -d '\n' > "$OPERATOR_SECRET_DIR/admin-token" +openssl rand -hex 32 | tr -d '\n' > "$OPERATOR_SECRET_DIR/viewer-token" +kubectl create namespace geaflow-system +kubectl -n geaflow-system create secret generic geaflow-operator-web-auth \ + --from-file=admin-token="$OPERATOR_SECRET_DIR/admin-token" \ + --from-file=viewer-token="$OPERATOR_SECRET_DIR/viewer-token" +``` + +The viewer key can be omitted to disable viewer access. The chart requires +`webSecurity.existingSecret` and `image.tag`; the administrator key is not optional. +Use an immutable, unique image tag built from the patched application, rather than reusing +a tag cached on cluster nodes. If you change Secret key names, set `adminTokenKey` and +`viewerTokenKey` under `webSecurity`. + +```sh +helm upgrade --install geaflow-kubernetes-operator \ + geaflow-kubernetes-operator/helm/geaflow-kubernetes-operator \ + --namespace geaflow-system \ + --set webSecurity.existingSecret=geaflow-operator-web-auth \ + --set image.repository=YOUR_PRIVATE_REGISTRY/geaflow-kubernetes-operator \ + --set image.tag=YOUR_UNIQUE_PATCHED_IMAGE_TAG \ + --set-json 'watchNamespaces=["geaflow-system"]' +``` + +The namespace setting above selects namespaced RBAC in the existing chart. It must match +this operator's current namespace: the reconciler watches the current namespace. Do not +interpret a list in `watchNamespaces` as configuring the reconciler to watch multiple +namespaces. Review effective RoleBindings and ClusterRoleBindings after upgrades, including +resources retained by prior releases. The chart's legacy cluster-scoped RBAC, wildcard +verbs and Secret permissions need a separate least-privilege review against enabled features. + +After transferring the tokens securely, delete the temporary files: + +```sh +rm -f "$OPERATOR_SECRET_DIR/admin-token" "$OPERATOR_SECRET_DIR/viewer-token" +rmdir "$OPERATOR_SECRET_DIR" +unset OPERATOR_SECRET_DIR +``` + +## Network and browser access + +The chart retains a ClusterIP service and disabled ingress. Its default NetworkPolicy denies +inbound pod traffic. A CNI that enforces NetworkPolicy is required; policies are additive, +so another policy may allow ingress. NetworkPolicy does not replace authentication. +The policy intentionally leaves operator egress unchanged. + +Use the Kubernetes-authenticated local tunnel for dashboard access: + +```sh +kubectl -n geaflow-system port-forward --address 127.0.0.1 \ + service/geaflow-kubernetes-operator 8089:80 +``` + +Open `http://127.0.0.1:8089/` and enter the viewer token for normal dashboard use. The dashboard +stores its token in page memory only, clears it on disconnect/reload or a 401 response, and +attaches it only to its own API URLs. Do not install untrusted scripts or browser extensions +on an administrator's browser. The static shell being accessible does not grant API access. + +For remote access, terminate HTTPS at a trusted ingress/proxy and protect the upstream +connection according to your cluster's trust boundary. Do not send bearer tokens over an +untrusted plaintext connection. If a proxy rewrites a URL prefix, serve the dashboard at a +prefix ending in `/` (or `/index.html`) and forward Authorization unchanged. Forwarded +identity headers alone never authenticate a request. Rate limiting and access logging, +with Authorization redacted, belong at the trusted gateway. + +To allow only a designated ingress controller, adapt both selectors and the port to your +cluster, and merge the following with your other Helm values. Supplying both selectors in +one `from` entry requires both to match: + +```yaml +networkPolicy: + enabled: true + ingress: + - from: + - namespaceSelector: + matchLabels: + kubernetes.io/metadata.name: ingress-nginx + podSelector: + matchLabels: + app.kubernetes.io/component: controller + ports: + - protocol: TCP + port: 8089 +``` + +API clients must send `Authorization: Bearer ` explicitly. Cookies, query-string +tokens, Basic authentication and sessions are not accepted. CORS is not enabled. CSRF is +disabled specifically because no browser-implicit authentication mechanism is enabled; +adding cookie/session/Basic authentication requires revisiting CSRF protection. + +## Rotation and deployment + +Update the externally managed Secret and restart every operator replica. Environment-backed +credentials do not reload when a Secret changes. Rollout completion, not Secret modification, +is the point when old credentials stop working. The chart applies its existing default +`Recreate` strategy, which entails downtime. If you select a rolling strategy, old and new +credentials may remain valid on different replicas during the transition. During the initial +upgrade from an unauthenticated version, isolate the service before deployment and keep it +isolated until every old pod has stopped. Confirm rollout success and readiness before +restoring approved access. Never restore an exposed unauthenticated image as a rollback. + +## Workload isolation + +Apply least-privilege operator and job service accounts, namespace isolation, Pod Security +Admission and appropriate resource quotas. Restrict executable images, service account +selection, host mounts, host networking and other deployment settings with admission policies +that cover generated pods/deployments and direct custom-resource submission. Do not let +untrusted jobs select the operator service account. An image-only allowlist does not make +user JARs, entry classes or arbitrary job settings safe. + +Review the resolved dependencies and the vendor support status before release. This targeted +change retains the existing Spring Boot 2.7 dependency baseline; it is not a framework upgrade. + +The web authentication change does not sandbox jobs, remove existing suspicious workloads, +revoke Kubernetes credentials, add admission policies, or establish per-user audit identity. +Investigate unexpected CRs, pods, deployments, service accounts and audit events separately. + +Report security issues privately using the [project security policy](../SECURITY.md). 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-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.boot spring-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..44cb75e51 --- /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,220 @@ +/* + * 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.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.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(path).header(HttpHeaders.AUTHORIZATION, "Bearer " + ADMIN)) + .andExpect(status().isBadRequest()); + } + 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); + } + } +} 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/NOTES.txt b/geaflow-kubernetes-operator/helm/geaflow-kubernetes-operator/templates/NOTES.txt index 3a69b793d..03fea68ec 100644 --- a/geaflow-kubernetes-operator/helm/geaflow-kubernetes-operator/templates/NOTES.txt +++ b/geaflow-kubernetes-operator/helm/geaflow-kubernetes-operator/templates/NOTES.txt @@ -1,22 +1,10 @@ -1. Get the application URL by running these commands: -{{- if .Values.ingress.enabled }} -{{- range $host := .Values.ingress.hosts }} - {{- range .paths }} - http{{ if $.Values.ingress.tls }}s{{ end }}://{{ $host.host }}{{ .path }} - {{- end }} -{{- end }} -{{- else if contains "NodePort" .Values.service.type }} - export NODE_PORT=$(kubectl get --namespace {{ .Release.Namespace }} -o jsonpath="{.spec.ports[0].nodePort}" services {{ include "geaflow-operator.fullname" . }}) - export NODE_IP=$(kubectl get nodes --namespace {{ .Release.Namespace }} -o jsonpath="{.items[0].status.addresses[0].address}") - echo http://$NODE_IP:$NODE_PORT -{{- else if contains "LoadBalancer" .Values.service.type }} - NOTE: It may take a few minutes for the LoadBalancer IP to be available. - You can watch the status of by running 'kubectl get --namespace {{ .Release.Namespace }} svc -w {{ include "geaflow-operator.fullname" . }}' - export SERVICE_IP=$(kubectl get svc --namespace {{ .Release.Namespace }} {{ include "geaflow-operator.fullname" . }} --template "{{"{{ range (index .status.loadBalancer.ingress 0) }}{{.}}{{ end }}"}}") - echo http://$SERVICE_IP:{{ .Values.service.port }} -{{- else if contains "ClusterIP" .Values.service.type }} - export POD_NAME=$(kubectl get pods --namespace {{ .Release.Namespace }} -l "app.kubernetes.io/name={{ include "geaflow-operator.name" . }},app.kubernetes.io/instance={{ .Release.Name }}" -o jsonpath="{.items[0].metadata.name}") - export CONTAINER_PORT=$(kubectl get pod --namespace {{ .Release.Namespace }} $POD_NAME -o jsonpath="{.spec.containers[0].ports[0].containerPort}") - echo "Visit http://127.0.0.1:8080 to use your application" - kubectl --namespace {{ .Release.Namespace }} port-forward $POD_NAME 8080:$CONTAINER_PORT -{{- end }} +The web API requires a bearer token from Secret {{ .Values.webSecurity.existingSecret }}. +Use a viewer token for read-only dashboard access. Tokens are never included in this output. + +Use a local Kubernetes-authenticated tunnel: + kubectl --namespace {{ .Release.Namespace }} port-forward --address 127.0.0.1 service/{{ include "geaflow-operator.fullname" . }} 8089:{{ .Values.service.port }} + +Open http://127.0.0.1:8089/ and enter the token in the dashboard. +The default NetworkPolicy denies ingress from other pods when supported by the CNI. +For remote access, configure HTTPS and explicit NetworkPolicy rules for approved clients. +See SECURITY_MODEL.md for deployment, permissions and credential rotation requirements. 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..adc122cc3 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 @@ -77,6 +79,17 @@ spec: value: "true" - 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 From 7592fa6a46807babad33e405b794490f7c7c1e6f Mon Sep 17 00:00:00 2001 From: kitalkuyo-gita Date: Fri, 30 Oct 2026 21:00:16 +0800 Subject: [PATCH 2/6] chore(operator): restore documentation changes --- SECURITY.md | 6 +- geaflow-kubernetes-operator/README.md | 24 +-- geaflow-kubernetes-operator/README_cn.md | 24 +-- geaflow-kubernetes-operator/SECURITY_MODEL.md | 168 ------------------ .../templates/NOTES.txt | 32 ++-- 5 files changed, 33 insertions(+), 221 deletions(-) delete mode 100644 geaflow-kubernetes-operator/SECURITY_MODEL.md diff --git a/SECURITY.md b/SECURITY.md index 3c638d9b9..02ddfd373 100644 --- a/SECURITY.md +++ b/SECURITY.md @@ -10,7 +10,7 @@ If you discover a security vulnerability in Apache GeaFlow (Incubating), please Please send your security vulnerability report to the Apache Security Team at: -**[security@apache.org](mailto:security@apache.org)** +**[dev@geaflow.apache.org](mailto:dev@geaflow.apache.org)** You can also report to the GeaFlow project team directly at: @@ -55,7 +55,3 @@ Security fixes will be released as part of regular GeaFlow releases and announce - [Apache Security Team](https://www.apache.org/security/) - [Apache GeaFlow Official Website](https://geaflow.apache.org/) - [Apache GeaFlow GitHub Repository](https://github.com/apache/geaflow) - -## Security Model - -- [Kubernetes operator security model and deployment requirements](geaflow-kubernetes-operator/SECURITY_MODEL.md) diff --git a/geaflow-kubernetes-operator/README.md b/geaflow-kubernetes-operator/README.md index 7bf0e52f5..56646d1fa 100644 --- a/geaflow-kubernetes-operator/README.md +++ b/geaflow-kubernetes-operator/README.md @@ -13,28 +13,14 @@ cd geaflow-kubernetes-operator ./build-operator.sh ``` 2. The image name and tag can be modified in helm/geaflow-kubernetes-operator/values. -3. Create the required authentication Secret following [the security model](SECURITY_MODEL.md). - Set a unique tag for an image built from this source. Then deploy: +3. Execute this command in the project root directory to quickly deploy to the kubernetes cluster. ```shell cd geaflow-kubernetes-operator -helm install geaflow-kubernetes-operator helm/geaflow-kubernetes-operator \ - --namespace geaflow-system \ - --set webSecurity.existingSecret=geaflow-operator-web-auth \ - --set image.tag=YOUR_UNIQUE_PATCHED_IMAGE_TAG \ - --set-json 'watchNamespaces=["geaflow-system"]' +helm install geaflow-kubernetes-operator helm/geaflow-kubernetes-operator ``` ### Frontend Dashboard -The dashboard requires an administrator or read-only bearer token. Prefer the read-only token. -The web service listens on port 8089 in Helm (8080 for standalone execution). The default -NetworkPolicy denies pod ingress; use a local tunnel and open `http://127.0.0.1:8089/`: - -```sh -kubectl -n geaflow-system port-forward --address 127.0.0.1 service/geaflow-kubernetes-operator 8089:80 -``` - -The dashboard displays job statistics and state. Tokens stay in page memory and must be -re-entered after reload. For TLS, credential rotation, RBAC and admission requirements, -read [the security model](SECURITY_MODEL.md). Job submission is reserved for trusted operators. +Geaflow-kubernetes-operator comes with a frontend dashboard page, which you can access through port 8089. +The dashboard will display job statistics within the cluster, as well as their current status and basic information. ### Example @@ -43,5 +29,5 @@ Please refer to example/example.yml and modify the necessary parameters. After h is completed, execute this command to submit the example job into your kubernetes cluster. ```shell cd geaflow-kubernetes-operator -kubectl -n geaflow-system apply -f example/example_hla.yml +kubectl apply -f example/example_hla.yml ``` diff --git a/geaflow-kubernetes-operator/README_cn.md b/geaflow-kubernetes-operator/README_cn.md index 2ddee031d..28f259180 100644 --- a/geaflow-kubernetes-operator/README_cn.md +++ b/geaflow-kubernetes-operator/README_cn.md @@ -12,34 +12,20 @@ cd geaflow-kubernetes-operator ./build-operator.sh ``` 2. 在helm/geaflow-kubernetes-operator/values中可以修改镜像名和tag。 -3. 先按[安全模型与部署要求](SECURITY_MODEL.md)在部署命名空间创建认证 Secret， - 并使用包含本次认证功能的独立镜像 tag，再执行部署： +3. 在项目根目录下执行以下命令即可快速部署到kubernetes集群中。 ```shell cd geaflow-kubernetes-operator -helm install geaflow-kubernetes-operator helm/geaflow-kubernetes-operator \ - --namespace geaflow-system \ - --set webSecurity.existingSecret=geaflow-operator-web-auth \ - --set image.tag=YOUR_UNIQUE_PATCHED_IMAGE_TAG \ - --set-json 'watchNamespaces=["geaflow-system"]' +helm install geaflow-kubernetes-operator helm/geaflow-kubernetes-operator ``` ### 前端Dashboard -Dashboard 需要管理员或只读 Bearer 令牌，日常查看优先使用只读令牌。 -Helm 中 Web 监听端口为 8089（独立启动默认 8080）。默认 NetworkPolicy 拒绝 Pod 入站访问， -可以建立本机隧道并访问 `http://127.0.0.1:8089/`： - -```sh -kubectl -n geaflow-system port-forward --address 127.0.0.1 service/geaflow-kubernetes-operator 8089:80 -``` - -Dashboard 展示作业统计、状态和基础信息。令牌只保存在页面内存，刷新后需重新输入。 -管理员提交作业等同于授权执行代码，只能交给可信运维人员。 -TLS、Secret 轮换、命名空间 RBAC 和准入策略要求见[安全模型](SECURITY_MODEL.md)。 +Geaflow-kubernetes-operator自带一个前端dashboard页面，你可以通过8089端口访问它。 +dashboard会展示集群内的作业统计，并展示它们的当前状态和基础信息。 ### 示例 请参考example/example.yml并修改必要参数。在helm部署完成后，执行以下命令提交示例的作业。 ```shell cd geaflow-kubernetes-operator -kubectl -n geaflow-system apply -f example/example_hla.yml +kubectl apply -f example/example_hla.yml ``` diff --git a/geaflow-kubernetes-operator/SECURITY_MODEL.md b/geaflow-kubernetes-operator/SECURITY_MODEL.md deleted file mode 100644 index f79f68010..000000000 --- a/geaflow-kubernetes-operator/SECURITY_MODEL.md +++ /dev/null @@ -1,168 +0,0 @@ -# Kubernetes operator security model - -This model describes the operator with mandatory web authentication. It does not apply -retroactively to older operator images. Installing this Helm chart around an old image -cannot add application authentication. - -## Trust and permissions - -The operator is a privileged workload deployment service. Only trusted cluster operators -may hold its administrator token or create/update GeaFlow custom resources directly. -Submitting a job means authorizing execution of its image, entry class and executable -artifacts, with the job's service account. Job configuration can influence deployment -settings as well. An administrator token is not a safe capability to give to untrusted -code authors or isolated tenants. - -The administrator token permits all web API methods. The optional, separate viewer token -permits only GET and HEAD under `/api`. Viewers can see all returned job specifications -and cluster information in the operator's namespace; those responses may contain sensitive -configuration. GET/HEAD endpoints must remain free of mutations. There is no per-job, -per-tenant or individual-user authorization. Tokens are shared deployment credentials, -not Kubernetes credentials, and requests are not impersonated into Kubernetes. - -Unauthenticated network clients, including other pods, must not read or mutate business -API resources. Only the GET health response and dashboard static files are public. -Missing, malformed, too-short or identical administrator/viewer credentials prevent startup. -Invalid/missing request credentials return HTTP 401; an authenticated viewer attempting a -write receives HTTP 403. No anonymous fallback or authentication-disable option is provided. - -## Configure credentials - -Generate independent random tokens with at least 32 random bytes (for example, -`openssl rand -hex 32`). The accepted representation is 32–512 characters using RFC 6750 -bearer-token characters. Length validation cannot establish entropy; do not use repeated -characters, example strings or passwords. - -For standalone execution set `GEAFLOW_OPERATOR_ADMIN_TOKEN` and optionally -`GEAFLOW_OPERATOR_VIEWER_TOKEN` in the process environment. The equivalent Spring properties -are `geaflow.operator.web.security.admin-token` and `geaflow.operator.web.security.viewer-token`. -Use a secret manager, protected environment or Spring configuration tree. Do not put -credentials in Git, Helm values, command-line arguments, URLs or application logs. - -For Kubernetes, create a Secret in the operator's namespace. This example keeps token values -out of command arguments and terminal output. Protect and distribute the files through your -approved secret-management process; the administrator and viewer tokens have different trust -requirements. Disable shell tracing before handling real tokens. - -```sh -set +x -umask 077 -OPERATOR_SECRET_DIR=$(mktemp -d) -openssl rand -hex 32 | tr -d '\n' > "$OPERATOR_SECRET_DIR/admin-token" -openssl rand -hex 32 | tr -d '\n' > "$OPERATOR_SECRET_DIR/viewer-token" -kubectl create namespace geaflow-system -kubectl -n geaflow-system create secret generic geaflow-operator-web-auth \ - --from-file=admin-token="$OPERATOR_SECRET_DIR/admin-token" \ - --from-file=viewer-token="$OPERATOR_SECRET_DIR/viewer-token" -``` - -The viewer key can be omitted to disable viewer access. The chart requires -`webSecurity.existingSecret` and `image.tag`; the administrator key is not optional. -Use an immutable, unique image tag built from the patched application, rather than reusing -a tag cached on cluster nodes. If you change Secret key names, set `adminTokenKey` and -`viewerTokenKey` under `webSecurity`. - -```sh -helm upgrade --install geaflow-kubernetes-operator \ - geaflow-kubernetes-operator/helm/geaflow-kubernetes-operator \ - --namespace geaflow-system \ - --set webSecurity.existingSecret=geaflow-operator-web-auth \ - --set image.repository=YOUR_PRIVATE_REGISTRY/geaflow-kubernetes-operator \ - --set image.tag=YOUR_UNIQUE_PATCHED_IMAGE_TAG \ - --set-json 'watchNamespaces=["geaflow-system"]' -``` - -The namespace setting above selects namespaced RBAC in the existing chart. It must match -this operator's current namespace: the reconciler watches the current namespace. Do not -interpret a list in `watchNamespaces` as configuring the reconciler to watch multiple -namespaces. Review effective RoleBindings and ClusterRoleBindings after upgrades, including -resources retained by prior releases. The chart's legacy cluster-scoped RBAC, wildcard -verbs and Secret permissions need a separate least-privilege review against enabled features. - -After transferring the tokens securely, delete the temporary files: - -```sh -rm -f "$OPERATOR_SECRET_DIR/admin-token" "$OPERATOR_SECRET_DIR/viewer-token" -rmdir "$OPERATOR_SECRET_DIR" -unset OPERATOR_SECRET_DIR -``` - -## Network and browser access - -The chart retains a ClusterIP service and disabled ingress. Its default NetworkPolicy denies -inbound pod traffic. A CNI that enforces NetworkPolicy is required; policies are additive, -so another policy may allow ingress. NetworkPolicy does not replace authentication. -The policy intentionally leaves operator egress unchanged. - -Use the Kubernetes-authenticated local tunnel for dashboard access: - -```sh -kubectl -n geaflow-system port-forward --address 127.0.0.1 \ - service/geaflow-kubernetes-operator 8089:80 -``` - -Open `http://127.0.0.1:8089/` and enter the viewer token for normal dashboard use. The dashboard -stores its token in page memory only, clears it on disconnect/reload or a 401 response, and -attaches it only to its own API URLs. Do not install untrusted scripts or browser extensions -on an administrator's browser. The static shell being accessible does not grant API access. - -For remote access, terminate HTTPS at a trusted ingress/proxy and protect the upstream -connection according to your cluster's trust boundary. Do not send bearer tokens over an -untrusted plaintext connection. If a proxy rewrites a URL prefix, serve the dashboard at a -prefix ending in `/` (or `/index.html`) and forward Authorization unchanged. Forwarded -identity headers alone never authenticate a request. Rate limiting and access logging, -with Authorization redacted, belong at the trusted gateway. - -To allow only a designated ingress controller, adapt both selectors and the port to your -cluster, and merge the following with your other Helm values. Supplying both selectors in -one `from` entry requires both to match: - -```yaml -networkPolicy: - enabled: true - ingress: - - from: - - namespaceSelector: - matchLabels: - kubernetes.io/metadata.name: ingress-nginx - podSelector: - matchLabels: - app.kubernetes.io/component: controller - ports: - - protocol: TCP - port: 8089 -``` - -API clients must send `Authorization: Bearer ` explicitly. Cookies, query-string -tokens, Basic authentication and sessions are not accepted. CORS is not enabled. CSRF is -disabled specifically because no browser-implicit authentication mechanism is enabled; -adding cookie/session/Basic authentication requires revisiting CSRF protection. - -## Rotation and deployment - -Update the externally managed Secret and restart every operator replica. Environment-backed -credentials do not reload when a Secret changes. Rollout completion, not Secret modification, -is the point when old credentials stop working. The chart applies its existing default -`Recreate` strategy, which entails downtime. If you select a rolling strategy, old and new -credentials may remain valid on different replicas during the transition. During the initial -upgrade from an unauthenticated version, isolate the service before deployment and keep it -isolated until every old pod has stopped. Confirm rollout success and readiness before -restoring approved access. Never restore an exposed unauthenticated image as a rollback. - -## Workload isolation - -Apply least-privilege operator and job service accounts, namespace isolation, Pod Security -Admission and appropriate resource quotas. Restrict executable images, service account -selection, host mounts, host networking and other deployment settings with admission policies -that cover generated pods/deployments and direct custom-resource submission. Do not let -untrusted jobs select the operator service account. An image-only allowlist does not make -user JARs, entry classes or arbitrary job settings safe. - -Review the resolved dependencies and the vendor support status before release. This targeted -change retains the existing Spring Boot 2.7 dependency baseline; it is not a framework upgrade. - -The web authentication change does not sandbox jobs, remove existing suspicious workloads, -revoke Kubernetes credentials, add admission policies, or establish per-user audit identity. -Investigate unexpected CRs, pods, deployments, service accounts and audit events separately. - -Report security issues privately using the [project security policy](../SECURITY.md). diff --git a/geaflow-kubernetes-operator/helm/geaflow-kubernetes-operator/templates/NOTES.txt b/geaflow-kubernetes-operator/helm/geaflow-kubernetes-operator/templates/NOTES.txt index 03fea68ec..3a69b793d 100644 --- a/geaflow-kubernetes-operator/helm/geaflow-kubernetes-operator/templates/NOTES.txt +++ b/geaflow-kubernetes-operator/helm/geaflow-kubernetes-operator/templates/NOTES.txt @@ -1,10 +1,22 @@ -The web API requires a bearer token from Secret {{ .Values.webSecurity.existingSecret }}. -Use a viewer token for read-only dashboard access. Tokens are never included in this output. - -Use a local Kubernetes-authenticated tunnel: - kubectl --namespace {{ .Release.Namespace }} port-forward --address 127.0.0.1 service/{{ include "geaflow-operator.fullname" . }} 8089:{{ .Values.service.port }} - -Open http://127.0.0.1:8089/ and enter the token in the dashboard. -The default NetworkPolicy denies ingress from other pods when supported by the CNI. -For remote access, configure HTTPS and explicit NetworkPolicy rules for approved clients. -See SECURITY_MODEL.md for deployment, permissions and credential rotation requirements. +1. Get the application URL by running these commands: +{{- if .Values.ingress.enabled }} +{{- range $host := .Values.ingress.hosts }} + {{- range .paths }} + http{{ if $.Values.ingress.tls }}s{{ end }}://{{ $host.host }}{{ .path }} + {{- end }} +{{- end }} +{{- else if contains "NodePort" .Values.service.type }} + export NODE_PORT=$(kubectl get --namespace {{ .Release.Namespace }} -o jsonpath="{.spec.ports[0].nodePort}" services {{ include "geaflow-operator.fullname" . }}) + export NODE_IP=$(kubectl get nodes --namespace {{ .Release.Namespace }} -o jsonpath="{.items[0].status.addresses[0].address}") + echo http://$NODE_IP:$NODE_PORT +{{- else if contains "LoadBalancer" .Values.service.type }} + NOTE: It may take a few minutes for the LoadBalancer IP to be available. + You can watch the status of by running 'kubectl get --namespace {{ .Release.Namespace }} svc -w {{ include "geaflow-operator.fullname" . }}' + export SERVICE_IP=$(kubectl get svc --namespace {{ .Release.Namespace }} {{ include "geaflow-operator.fullname" . }} --template "{{"{{ range (index .status.loadBalancer.ingress 0) }}{{.}}{{ end }}"}}") + echo http://$SERVICE_IP:{{ .Values.service.port }} +{{- else if contains "ClusterIP" .Values.service.type }} + export POD_NAME=$(kubectl get pods --namespace {{ .Release.Namespace }} -l "app.kubernetes.io/name={{ include "geaflow-operator.name" . }},app.kubernetes.io/instance={{ .Release.Name }}" -o jsonpath="{.items[0].metadata.name}") + export CONTAINER_PORT=$(kubectl get pod --namespace {{ .Release.Namespace }} $POD_NAME -o jsonpath="{.spec.containers[0].ports[0].containerPort}") + echo "Visit http://127.0.0.1:8080 to use your application" + kubectl --namespace {{ .Release.Namespace }} port-forward $POD_NAME 8080:$CONTAINER_PORT +{{- end }} From b3909dcb7f271df0ddc5ee7bd7656e543730e6e6 Mon Sep 17 00:00:00 2001 From: kitalkuyo-gita Date: Sat, 31 Oct 2026 09:36:34 +0800 Subject: [PATCH 3/6] fix(ci): stabilize Docker-backed connector tests --- .github/workflows/ci-jdk11.yml | 8 +- .github/workflows/ci.yml | 8 +- .../geaflow-dsl-connector-clickhouse/pom.xml | 15 +- ...ickHouseTableConnectorIntegrationTest.java | 244 ++++++++++++++++++ .../ClickHouseTableConnectorTest.java | 209 +-------------- .../clickhouse/ClickHouseTableSinkTest.java | 115 +++++++++ .../clickhouse/ClickHouseTestEnvironment.java | 64 +++++ .../ClickHouseTestEnvironmentTest.java | 86 ++++++ .../src/test/resources/log4j2-test.xml | 33 +++ .../geaflow-dsl-connector-doris/pom.xml | 1 - .../geaflow-dsl-connector-mongodb/pom.xml | 1 - .../geaflow-dsl/geaflow-dsl-connector/pom.xml | 5 + tools/ci/check-docker.sh | 36 +++ tools/ci/test_check_docker.py | 75 ++++++ 14 files changed, 687 insertions(+), 213 deletions(-) create mode 100644 geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-clickhouse/src/test/java/org/apache/geaflow/dsl/connector/clickhouse/ClickHouseTableConnectorIntegrationTest.java create mode 100644 geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-clickhouse/src/test/java/org/apache/geaflow/dsl/connector/clickhouse/ClickHouseTableSinkTest.java create mode 100644 geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-clickhouse/src/test/java/org/apache/geaflow/dsl/connector/clickhouse/ClickHouseTestEnvironment.java create mode 100644 geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-clickhouse/src/test/java/org/apache/geaflow/dsl/connector/clickhouse/ClickHouseTestEnvironmentTest.java create mode 100644 geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-clickhouse/src/test/resources/log4j2-test.xml create mode 100644 tools/ci/check-docker.sh create mode 100644 tools/ci/test_check_docker.py diff --git a/.github/workflows/ci-jdk11.yml b/.github/workflows/ci-jdk11.yml index e5878aaa2..aedbad0a6 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: @@ -79,4 +85,4 @@ jobs: run: | test_modules="!geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-hive," test_modules+="!geaflow/geaflow-dsl/geaflow-dsl-connector-tests" - mvn -B -e clean test -Pjdk11 -pl "${test_modules}" -Duser.timezone=Asia/Shanghai -Dlog4j.configuration="log4j.rootLogger=WARN, stdout" + mvn -B -e clean test -Pjdk11 -pl "${test_modules}" -Dclickhouse.it.required=true -Duser.timezone=Asia/Shanghai -Dlog4j.configuration="log4j.rootLogger=WARN, stdout" diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index bba2534f6..4a7af377b 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: @@ -81,5 +87,5 @@ jobs: run: | mvn -B -e clean test -Pjdk8 \ -pl !geaflow/geaflow-plugins/geaflow-store/geaflow-store-vector \ - -Duser.timezone=Asia/Shanghai \ + -Dclickhouse.it.required=true -Duser.timezone=Asia/Shanghai \ -Dlog4j.configuration="log4j .rootLogger=WARN, stdout" 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.5 5.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() From f148721ed72e27bbdb49a096b99a95a9afd6b128 Mon Sep 17 00:00:00 2001 From: kitalkuyo-gita Date: Sat, 31 Oct 2026 10:58:37 +0800 Subject: [PATCH 4/6] test(ai): align keyword golden tests with store lifecycle Use finishWriting before querying or inspecting the index and close stores during cleanup. Assert terminal close behavior and Directory release while preserving the existing tokenization and serialized-document golden data. --- .../KeywordIndexLifecycleGoldenTest.java | 148 ++++++++++-------- .../KeywordIndexSerializationGoldenTest.java | 56 ++++--- 2 files changed, 116 insertions(+), 88 deletions(-) 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 { From 7ca6281a3a1431c2aa11d16b8d7dc822e0d1d292 Mon Sep 17 00:00:00 2001 From: kitalkuyo-gita Date: Sat, 31 Oct 2026 15:54:21 +0800 Subject: [PATCH 5/6] ci: remove redundant ClickHouse required flag --- .github/workflows/ci-jdk11.yml | 2 +- .github/workflows/ci.yml | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/.github/workflows/ci-jdk11.yml b/.github/workflows/ci-jdk11.yml index aedbad0a6..26e0991b6 100644 --- a/.github/workflows/ci-jdk11.yml +++ b/.github/workflows/ci-jdk11.yml @@ -85,4 +85,4 @@ jobs: run: | test_modules="!geaflow/geaflow-dsl/geaflow-dsl-connector/geaflow-dsl-connector-hive," test_modules+="!geaflow/geaflow-dsl/geaflow-dsl-connector-tests" - mvn -B -e clean test -Pjdk11 -pl "${test_modules}" -Dclickhouse.it.required=true -Duser.timezone=Asia/Shanghai -Dlog4j.configuration="log4j.rootLogger=WARN, stdout" + mvn -B -e clean test -Pjdk11 -pl "${test_modules}" -Duser.timezone=Asia/Shanghai -Dlog4j.configuration="log4j.rootLogger=WARN, stdout" diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 4a7af377b..217afc9c7 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -87,5 +87,5 @@ jobs: run: | mvn -B -e clean test -Pjdk8 \ -pl !geaflow/geaflow-plugins/geaflow-store/geaflow-store-vector \ - -Dclickhouse.it.required=true -Duser.timezone=Asia/Shanghai \ + -Duser.timezone=Asia/Shanghai \ -Dlog4j.configuration="log4j .rootLogger=WARN, stdout" From 1662c9135e910c2ff44934b46ee81604dd4b9c37 Mon Sep 17 00:00:00 2001 From: kitalkuyo-gita Date: Sat, 31 Oct 2026 17:23:17 +0800 Subject: [PATCH 6/6] fix(operator): repair standalone builds and container startup Inherit Spring Boot test dependency management, align the engine and Checkstyle versions, remove the unsupported TestNG provider, and complete Kubernetes imports and security-test fixtures. Build on a pinned Java 11 runtime and run the JVM in the foreground so invalid credentials stop the container. Align Helm's local-mode environment with the entrypoint. --- geaflow-kubernetes-operator/Dockerfile | 25 ++--------- .../docker/bin/start-process.sh | 42 ++++++------------- .../operator/core/util/KubernetesUtil.java | 8 ++++ .../web/security/WebSecurityTest.java | 11 ++++- .../templates/deployment.yaml | 4 +- geaflow-kubernetes-operator/pom.xml | 21 ++-------- 6 files changed, 38 insertions(+), 73 deletions(-) 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-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/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 index 44cb75e51..f41630c3b 100644 --- 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 @@ -35,10 +35,12 @@ 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; @@ -192,8 +194,8 @@ public void exposesHealthButDoesNotExemptOtherHealthPathsOrUnprefixedControllers @Test public void rejectsNonCanonicalPathsBeforeControllerDispatch() throws Exception { for (String path : new String[] {"/api//jobs", "/api/./jobs", "/api/jobs;x=1"}) { - mvc.perform(get(path).header(HttpHeaders.AUTHORIZATION, "Bearer " + ADMIN)) - .andExpect(status().isBadRequest()); + mvc.perform(get(URI.create(path)).header(HttpHeaders.AUTHORIZATION, "Bearer " + ADMIN)) + .andExpect(result -> assertEquals(path, 400, result.getResponse().getStatus())); } verifyNoInteractions(service); } @@ -216,5 +218,10 @@ static class TestConfiguration { GeaflowJobApiService geaflowJobApiService() { return mock(GeaflowJobApiService.class); } + + @Bean + JobCRCache jobCRCache() { + return mock(JobCRCache.class); + } } } 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 adc122cc3..c29e1324a 100644 --- a/geaflow-kubernetes-operator/helm/geaflow-kubernetes-operator/templates/deployment.yaml +++ b/geaflow-kubernetes-operator/helm/geaflow-kubernetes-operator/templates/deployment.yaml @@ -75,8 +75,8 @@ 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 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.2 2.0.29 20.0 + 0.8.0-SNAPSHOT 11 11 11 @@ -107,12 +108,6 @@ ${junit.version} - - org.springframework.boot - spring-boot-starter-test - test - - org.apache.geaflow geaflow-kubernetes-operator-bootstrap @@ -146,7 +141,7 @@ org.apache.geaflow geaflow-on-k8s - ${project.version} + ${geaflow.version} @@ -157,12 +152,12 @@ org.apache.maven.plugins maven-checkstyle-plugin - 2.17 + 3.3.0 checkstyle com.puppycrawl.tools - 7.1 + 8.45 @@ -212,11 +207,6 @@ org.apache.maven.surefire 2.20 - - surefire-testng - org.apache.maven.surefire - 2.20 - @@ -226,9 +216,6 @@ **/*Tests.java **/*Test.java - - false - 1