diff --git a/.github/actions/helm-checks/action.yml b/.github/actions/helm-checks/action.yml index b5a596d524..344e4655dd 100644 --- a/.github/actions/helm-checks/action.yml +++ b/.github/actions/helm-checks/action.yml @@ -38,11 +38,17 @@ runs: helm lint helm/hugegraph -f helm/hugegraph/values-single.yaml helm lint helm/hugegraph -f helm/hugegraph/values-cluster.yaml helm lint helm/hugegraph -f helm/hugegraph/values-strict.yaml + helm lint helm/hugegraph -f helm/hugegraph/examples/values-seatunnel-standalone.yaml + helm lint helm/hugegraph -f helm/hugegraph/examples/values-seatunnel-cluster.yaml - name: packaged chart regression shell: 'bash -e {0}' run: python3 -m unittest discover -s helm -p 'test_helm_package.py' -v + - name: SeaTunnel integration regression + shell: 'bash -e {0}' + run: python3 -m unittest discover -s helm -p 'test_helm_seatunnel.py' -v + - name: helm unittest shell: 'bash -e {0}' run: | @@ -56,7 +62,8 @@ runs: - name: helm template shell: 'bash -e {0}' run: | - for preset in "" "-f helm/hugegraph/values-single.yaml" "-f helm/hugegraph/values-cluster.yaml" "-f helm/hugegraph/values-strict.yaml"; do + for preset in "" "-f helm/hugegraph/values-single.yaml" "-f helm/hugegraph/values-cluster.yaml" "-f helm/hugegraph/values-strict.yaml" \ + "-f helm/hugegraph/examples/values-seatunnel-standalone.yaml" "-f helm/hugegraph/examples/values-seatunnel-cluster.yaml"; do # shellcheck disable=SC2086 # $preset intentionally splits into flags helm template ci helm/hugegraph $preset > /dev/null done @@ -493,7 +500,8 @@ runs: curl -sSLo /tmp/kc.tar.gz https://github.com/yannh/kubeconform/releases/download/v0.6.7/kubeconform-linux-amd64.tar.gz echo "95f14e87aa28c09d5941f11bd024c1d02fdc0303ccaa23f61cef67bc92619d73 /tmp/kc.tar.gz" | sha256sum -c - tar -xzf /tmp/kc.tar.gz -C /tmp - for preset in "" "-f helm/hugegraph/values-single.yaml" "-f helm/hugegraph/values-cluster.yaml" "-f helm/hugegraph/values-strict.yaml"; do + for preset in "" "-f helm/hugegraph/values-single.yaml" "-f helm/hugegraph/values-cluster.yaml" "-f helm/hugegraph/values-strict.yaml" \ + "-f helm/hugegraph/examples/values-seatunnel-standalone.yaml" "-f helm/hugegraph/examples/values-seatunnel-cluster.yaml"; do # shellcheck disable=SC2086 # $preset intentionally splits into flags helm template ci helm/hugegraph $preset | /tmp/kubeconform -strict -summary -kubernetes-version 1.23.0 done diff --git a/LICENSE b/LICENSE index 8445ec58dc..105c58618a 100644 --- a/LICENSE +++ b/LICENSE @@ -218,3 +218,4 @@ hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/opencypher/Cyp hugegraph-server/hugegraph-api/src/main/java/org/apache/hugegraph/opencypher/CypherPlugin.java from https://github.com/opencypher/cypher-for-gremlin hugegraph-struct/src/main/java/org/apache/hugegraph/type/define/Cardinality.java from https://github.com/JanusGraph/janusgraph hugegraph-struct/src/main/java/org/apache/hugegraph/type/Namifiable.java from https://github.com/JanusGraph/janusgraph +helm/hugegraph/templates/seatunnel.yaml adapts the native hybrid Kubernetes deployment from Apache SeaTunnel 3.0.0 (Apache License 2.0). diff --git a/NOTICE b/NOTICE index 8e48b813b8..d770d1475e 100644 --- a/NOTICE +++ b/NOTICE @@ -6,6 +6,10 @@ The Apache Software Foundation (http://www.apache.org/). The initial codebase was donated to the ASF by HugeGraph Authors, copyright 2017-2021. +The optional native Kubernetes deployment in helm/hugegraph/templates/seatunnel.yaml adapts resources from: +Apache SeaTunnel +Copyright 2021-2024 The Apache Software Foundation + ----------------------------------------------------------------------- This product contains code form the Apache TinkerPop Project: diff --git a/helm/hugegraph/Chart.yaml b/helm/hugegraph/Chart.yaml index 16397760b1..e38988e2a7 100644 --- a/helm/hugegraph/Chart.yaml +++ b/helm/hugegraph/Chart.yaml @@ -17,9 +17,9 @@ apiVersion: v2 name: hugegraph-helm -description: Helm chart for Apache HugeGraph HStore cluster (PD + Store + Server) +description: Helm chart for Apache HugeGraph HStore cluster with optional SeaTunnel type: application -version: 0.2.0 +version: 0.3.0 # TODO: set appVersion to "1.8.0" when the release images are published. appVersion: "latest" kubeVersion: ">=1.23.0-0" @@ -32,6 +32,7 @@ home: https://hugegraph.apache.org/ icon: https://hugegraph.apache.org/img/logo.svg sources: - https://github.com/apache/hugegraph + - https://github.com/apache/seatunnel maintainers: - name: HugeGraph Community url: https://hugegraph.apache.org/ diff --git a/helm/hugegraph/README.md b/helm/hugegraph/README.md index e771e09c9e..597beebd77 100644 --- a/helm/hugegraph/README.md +++ b/helm/hugegraph/README.md @@ -81,6 +81,28 @@ helm install hugegraph ./helm/hugegraph -n hugegraph --create-namespace \ Inactive credential values may remain in the values file when auth is disabled. +### Optional SeaTunnel + +Set `seatunnel.enabled=true` to start one persistent hybrid member alongside HugeGraph; +set `seatunnel.mode=cluster` to use multiple members of the same native StatefulSet +(two by default). SeaTunnel is disabled by default and its topology is independent +of the HugeGraph single-/multi-node presets. Both modes include the engine's task UI. +The HTTP Service inherits NodePort in `easy` mode or ClusterIP in `strict` mode; +`seatunnel.service` can explicitly override the type and NodePort. + +Member/CLI port 5801 is restricted to same-release peers by a SeaTunnel ingress +NetworkPolicy in every mode, including when the chart-wide policy flag is disabled. +Easy mode leaves the authenticated HTTP entry open. Actual isolation requires CNI +support and trusted namespace RBAC; full isolation additionally limits egress. + +The chart injects the same-release Server address, GraphSpace and administrator +Secret references into the mounted Source/Sink examples. Create the target graph +before submitting them manually; Helm does not create graphs or import data. +UI and REST submission use a separate native HTTP Basic administrator account, +whose generated Secret is retained and reused during online upgrades. Container +security and scheduling options follow the other components; see the +[SeaTunnel deployment guide](docs/seatunnel.md) for configuration and limits. + ### Internal image registry Mirror the four component images (`hugegraph/pd`, `hugegraph/store`, diff --git a/helm/hugegraph/docs/seatunnel.md b/helm/hugegraph/docs/seatunnel.md new file mode 100644 index 0000000000..14a209a2c2 --- /dev/null +++ b/helm/hugegraph/docs/seatunnel.md @@ -0,0 +1,359 @@ +# Deploy HugeGraph with SeaTunnel + +This chart adapts the [SeaTunnel 3.0.0 native hybrid Kubernetes deployment](https://github.com/apache/seatunnel/blob/v3.0.0/docs/en/getting-started/kubernetes/hybrid-cluster-mode.md) +for HugeGraph resource names, DNS, images, configuration and replica counts. +It does not introduce a subchart, operator, CRD or derived SeaTunnel image. + +SeaTunnel is disabled by default. When enabled, a persistent hybrid member both +schedules and executes jobs. Cluster mode scales the same StatefulSet to multiple +members; it is not a `local-mode` Pod or Job that exits after one submission. + +| Setting | Meaning | +|---|---| +| `values-single.yaml` | HugeGraph 1 PD + 1 Store + 1 Server using HStore | +| `values-cluster.yaml` | HugeGraph's own multi-replica topology | +| `seatunnel.mode: standalone` | One persistent hybrid member | +| `seatunnel.mode: cluster`, `replicas: 2` | Two hybrid members; at least two are required | + +Choose HugeGraph and SeaTunnel topologies independently. Top-level `mode: easy/strict` +controls HTTP entry and full NetworkPolicy defaults. Both SeaTunnel modes provide +the same built-in UI and API. + +## Install, access and switch modes + +From the HugeGraph repository root, choose one installation: + +```bash +# HugeGraph 1 PD + 1 Store + 1 Server, plus one persistent SeaTunnel member +helm install hugegraph ./helm/hugegraph -n hugegraph --create-namespace \ + -f helm/hugegraph/values-single.yaml \ + -f helm/hugegraph/examples/values-seatunnel-standalone.yaml \ + --wait --timeout 30m + +# The same HugeGraph topology, plus two SeaTunnel hybrid members +helm install hugegraph ./helm/hugegraph -n hugegraph --create-namespace \ + -f helm/hugegraph/values-single.yaml \ + -f helm/hugegraph/examples/values-seatunnel-cluster.yaml \ + --wait --timeout 30m +``` + +The default image is `apache/seatunnel:3.0.0`. Both modes expose the UI and REST +API on HTTP 8080. Easy mode inherits a NodePort, while strict mode inherits +ClusterIP. Override `seatunnel.service.type/nodePort` explicitly when needed. +`helm install` prints access instructions; retrieve an automatically allocated +HTTP NodePort with: + +```bash +kubectl -n hugegraph get svc hugegraph-seatunnel \ + -o jsonpath='{.spec.ports[?(@.name=="http")].nodePort}{"\n"}' +``` + +For any Service type, port forwarding provides a local entry: + +```bash +kubectl -n hugegraph port-forward svc/hugegraph-seatunnel 8081:8080 +``` + +Open and use the separate SeaTunnel credentials +described below. The UI shows cluster members, jobs, DAGs and logs. Submit jobs +through the CLI or authenticated REST API. Helm starts the engine and prepares +examples; it does not create graphs or run jobs. + +Only HTTP 8080 is exposed through NodePort or LoadBalancer. A separate ingress +NetworkPolicy always restricts Hazelcast member/CLI port 5801 to same-release +SeaTunnel peers, even in easy mode or with `networkPolicy.enabled=false`. +Easy mode permits HTTP callers without adding an ingress rule and leaves egress +unrestricted. When full NetworkPolicy isolation is enabled, explicitly allow +HTTP callers and external source/sink destinations with +`networkPolicy.seatunnel.extraIngress/extraEgress`. Explicit extra ingress rules +can also broaden member access, so restrict them to trusted callers. + +Actual isolation requires a NetworkPolicy-capable CNI. Pod selectors are not +native Hazelcast authentication: use trusted namespace RBAC/admission controls +for creating or relabeling Pods and for `kubectl exec`. Do not expose 5801 outside +that administrative boundary. The chart does not manage a separate SeaTunnel Ingress. + +For release `hugegraph`, resources are the `hugegraph-seatunnel` StatefulSet and +HTTP Service, `hugegraph-seatunnel-headless` discovery Service, and +`hugegraph-seatunnel-configs` ConfigMap. Names honor `fullnameOverride`. +Members use Headless Service DNS rather than the Kubernetes API, so discovery +requires no Kubernetes API token or RBAC permission. + +Stop all jobs before switching modes. Retain the complete installation values +and upgrade with: + +```bash +helm upgrade hugegraph ./helm/hugegraph -n hugegraph -f my-values.yaml \ + --set seatunnel.enabled=true --set seatunnel.mode=cluster \ + --set seatunnel.replicas=2 --wait --timeout 30m +``` + +Set `seatunnel.mode=standalone` to return to one member. The StatefulSet stays the +same, but replica count and engine configuration change. Uninterrupted switching, +automatic job recovery and resubmission are not guaranteed. Setting +`seatunnel.enabled=false` removes the engine workloads and Services without +deleting data already written to HugeGraph. The retained auth Secret can be +reused after re-enabling or reinstalling. + +## SeaTunnel entry authentication + +The UI, static resources and REST submission endpoints use native HTTP Basic +authentication. There is no added proxy or login service. `seatunnel.auth.username` +defaults to `admin`. An empty password generates a strong password in +`-auth`; online upgrades reuse the existing password. + +Use a controlled values file to set or rotate `seatunnel.auth.username/password`. +Values cannot contain leading/trailing whitespace or `${...}` placeholders, and +usernames cannot contain a colon. Internal spaces and quotes are accepted. +Pin `seatunnel.auth.password` for offline templates or GitOps rendering; automatic +password reuse requires an online Helm lookup. + +The default release uses Secret `hugegraph-seatunnel-auth`. With port forwarding +active, read its credentials into local variables and call REST: + +```bash +ST_USERNAME="$(kubectl -n hugegraph get secret hugegraph-seatunnel-auth \ + -o jsonpath='{.data.username}' | base64 --decode)" +ST_PASSWORD="$(kubectl -n hugegraph get secret hugegraph-seatunnel-auth \ + -o jsonpath='{.data.password}' | base64 --decode)" + +curl -fsS --user "$ST_USERNAME:$ST_PASSWORD" http://127.0.0.1:8081/overview +``` + +Use the same credentials for the browser Basic login and any NodePort or +LoadBalancer entry. External job submissions must authenticate to the protected +HTTP entry using the [official REST API](https://github.com/apache/seatunnel/blob/v3.0.0/docs/en/engines/zeta/rest-api-and-web-ui.md). + +The Secret contains `username`, `password` and the final `seatunnel.yaml`. +The engine configuration is mounted directly from the Secret, not embedded in +a ConfigMap, Java arguments or logs. Rotate through controlled values and +`helm upgrade`, which updates the Secret and rolls the engine Pods. Editing only +`data.password` does not update credentials inside the engine configuration. +`HG_USERNAME/HG_PASSWORD` are separate HugeGraph connector credentials and cannot +replace entry authentication. Disabling `server.auth.enabled` does not disable +SeaTunnel authentication. + +HTTP Basic does not authenticate the Hazelcast member/CLI protocol on 5801. +The examples below execute the CLI inside an engine Pod under Kubernetes RBAC; +external clients should use REST. This is a single administrator entry, not +per-user/per-job authorization or HugeGraph RBAC/SSO. Use an existing TLS entry +when exposing the HTTP service outside a trusted network. + +## Images, credentials and native configuration + +The official `apache/seatunnel:3.0.0` image used by this chart includes the +HugeGraph, FakeSource and Console connectors and the built-in UI. Verify plugin +availability when changing the version or image. Configure +`seatunnel.image.repository/tag/pullPolicy`; it inherits `global.imageRegistry` +and `imagePullSecrets`. Helm does not download plugins at startup. The upstream +image contains many connectors, so allow sufficient pull time or pre-pull/mirror +it into an internal registry. + +The chart provides these defaults for same-release HugeGraph example jobs: + +| Variable | Default source | +|---|---| +| `HG_HOST` | Same-release HugeGraph Server Service DNS | +| `HG_PORT` | `server.port` | +| `HG_PROTOCOL` | `http` for the internal Server connection | +| `HG_USERNAME` | `admin` | +| `HG_PASSWORD` | Actual Server administrator Secret name/key, or empty when auth is disabled | +| `HG_GRAPH_SPACE` | `server.graphSpace` | +| `HG_GRAPH_NAME` | `graph_a` | + +`HG_PASSWORD` uses a native `secretKeyRef`, including +`server.auth.admin.existingSecret` and custom keys. It is not copied into example +job files. The chart does not create an extra HugeGraph account. Follow the +Server account synchronization procedure after rotating its credential Secret, +then restart relevant engine Pods to refresh their environment. + +A matching name in `seatunnel.env` overrides an automatic value without creating +duplicate entries. For example, select another graph and an existing job account: + +```yaml +seatunnel: + env: + - name: HG_GRAPH_NAME + value: import_demo + - name: HG_USERNAME + value: seatunnel-job + - name: HG_PASSWORD + valueFrom: + secretKeyRef: + name: hugegraph-job-credentials + key: password +``` + +The account must already exist and have the required data/Schema permissions. +These defaults serve only the local Source/Sink examples. Cross-environment jobs +require the explicit `HG_SOURCE_*` and `HG_TARGET_*` values below; the chart does +not discover remote graphs. Connector credentials do not replace entry auth. + +Tune the JVM with `seatunnel.javaOpts` and container resources with +`seatunnel.resources`. Defaults retain G1 and OOM heap dumps, with a 1 GiB maximum +heap and 256 MiB metaspace cap to leave room for native memory under the 2 GiB limit. +Heap dumps use writable `/tmp/seatunnel.hprof` and are ephemeral. `javaOpts` and +`config.jvm_options` each replace the complete image JVM options file; keep the +required GC and diagnostic flags when overriding either. `seatunnel.podSecurityContext` and +`seatunnel.securityContext` follow the other chart components. The container +starts with privilege escalation disabled, all capabilities dropped and +`RuntimeDefault` seccomp. The upstream image runs as root: configure a compatible +UID and writable paths when enforcing `runAsNonRoot` or a restricted Pod Security +namespace; these defaults alone do not establish compatibility with that policy. +Use `seatunnel.nodeSelector`, `tolerations`, `affinity` and `priorityClassName` for +node placement and scheduling. + +Override complete native configuration files through `seatunnel.config`, for example: + +```yaml +seatunnel: + config: + jvm_client_options: | + -Xms128m + -Xmx256m +``` + +The ConfigMap supplies discovery, JVM and example job files. The authenticated +`seatunnel.yaml` is mounted from its separate Secret; other files use the image +defaults. Final HTTP Basic settings remain managed by `seatunnel.auth` even when +overriding engine configuration. Retain member port 5801, HTTP 8080 and +`enable-http: true` so Services, probes and UI remain usable. Checkpoints use an +ephemeral local directory; this integration does not guarantee durable recovery. +Full NetworkPolicy isolation allows peer traffic, DNS and the same-release +Server API by default. See the [Chart README](../README.md) for HugeGraph access, +credentials and storage. + +## Same-release examples: import and read + +The chart mounts [hugegraph-sink.conf](../examples/seatunnel/hugegraph-sink.conf) +and [hugegraph-source.conf](../examples/seatunnel/hugegraph-source.conf) as +`/opt/seatunnel/config/hugegraph-sink.conf` and +`/opt/seatunnel/config/hugegraph-source.conf`. The Sink writes ten fixed vertices +with label `seatunnel_demo_person`; the Source reads that label into Console. +Both use the automatic `HG_*` environment values. Override a file under the same +name in `seatunnel.config` to customize a job. + +Create `graph_a` through Hubble or the HugeGraph API first, or create the graph +selected by explicit `HG_GRAPH_SPACE/HG_GRAPH_NAME` values. The default release, +namespace and GraphSpace are used below. Local `HG_USERNAME/HG_PASSWORD` must +come from your credential manager; Pod environment values do not reach your shell. + +```bash +# Keep Server port forwarding active in a separate terminal +kubectl -n hugegraph port-forward svc/hugegraph-server 8082:8080 +``` + +```bash +export HG_URL=http://127.0.0.1:8082 +curl -fsS -u "$HG_USERNAME:$HG_PASSWORD" -H 'Content-Type: application/json' \ + -X POST "$HG_URL/graphspaces/DEFAULT/graphs/graph_a" -d '{}' +``` + +Do not recreate an existing graph. Submit the mounted Sink, then Source: + +```bash +kubectl -n hugegraph exec sts/hugegraph-seatunnel -- \ + /opt/seatunnel/bin/seatunnel.sh --master cluster \ + --config /opt/seatunnel/config/hugegraph-sink.conf + +kubectl -n hugegraph exec sts/hugegraph-seatunnel -- \ + /opt/seatunnel/bin/seatunnel.sh --master cluster \ + --config /opt/seatunnel/config/hugegraph-source.conf +``` + +`--master cluster` submits to the persistent engine, including standalone mode. +On an initially empty graph, expect a finished Sink with ten rows read/written, +and a Source with ten rows read into the executing member's log. Inspect UI +status/DAG, all member logs and the actual graph data: + +```bash +kubectl -n hugegraph logs -l app.kubernetes.io/instance=hugegraph,app.kubernetes.io/component=seatunnel \ + --all-containers --prefix --tail=200 + +curl -fsS -u "$HG_USERNAME:$HG_PASSWORD" -H 'Content-Type: application/json' \ + "$HG_URL/gremlin" -d '{"gremlin":"graph.traversal().V().hasLabel(\"seatunnel_demo_person\").count()","aliases":{"graph":"DEFAULT-graph_a"}}' +``` + +The count query should return `result.data: [10]`. A cluster job can execute on +any member, so Console output is not limited to `hugegraph-seatunnel-0`. Neither +example runs automatically during Helm installation or upgrade. + +## Cross-environment copy: Schema, vertices and edges + +Use an existing source graph A and an empty graph B in another environment. +Copy **Schema, then vertices, then edges**. Sink auto-schema mapping cannot +replace copying the complete source Schema, indexes and IDs. + +Create `graph_b` through the native API, then run +[copy-schema.py](../examples/seatunnel/clone/copy-schema.py). It copies property +keys, vertex labels, edge labels and indexes with source Schema IDs while the +target is in RESTORING mode. It does not copy graph data. Use compatible HugeGraph +versions and accounts with administrative permissions; target Schema must be empty. + +Only NORMAL edge labels are supported. The script rejects PARENT, SUB or unknown +edge types before contacting the target; inherited Schema needs a separate +migration procedure. + +```bash +# Both URLs include GraphSpace and graph; credentials come from your local environment +export HG_SOURCE_GRAPH_URL=https://source.example.com/graphspaces/DEFAULT/graphs/graph_a +export HG_TARGET_GRAPH_URL=https://target.example.com/graphspaces/DEFAULT/graphs/graph_b +python3 helm/hugegraph/examples/seatunnel/clone/copy-schema.py \ + --source "$HG_SOURCE_GRAPH_URL" --target "$HG_TARGET_GRAPH_URL" +``` + +The script reads `HG_SOURCE_USERNAME/HG_SOURCE_PASSWORD` and +`HG_TARGET_USERNAME/HG_TARGET_PASSWORD`. Configure corresponding credentials in +the engine Pods through Secrets, plus `HG_SOURCE_HOST/PORT/PROTOCOL` and +`HG_TARGET_HOST/PORT/PROTOCOL`. Addresses must be reachable from the engine Pods, +not merely local port-forward addresses. + +The four example jobs read `graph_a` and write `graph_b`, copying both vertex +labels before either edge label: + +```bash +for clone_job in person company works_at knows; do + kubectl -n hugegraph exec -i sts/hugegraph-seatunnel -- \ + sh -c "cat > /tmp/$clone_job.conf" \ + < "helm/hugegraph/examples/seatunnel/clone/$clone_job.conf" + kubectl -n hugegraph exec sts/hugegraph-seatunnel -- \ + /opt/seatunnel/bin/seatunnel.sh --master cluster --config "/tmp/$clone_job.conf" \ + || break +done +``` + +Mappings use `ERROR_WHEN_SCHEMA_NOT_EXIST` and the already copied Schema. Edges +require their endpoint vertices first. See [person](../examples/seatunnel/clone/person.conf), +[company](../examples/seatunnel/clone/company.conf), [works_at](../examples/seatunnel/clone/works_at.conf) +and [knows](../examples/seatunnel/clone/knows.conf). The official Sink requires +explicit `mappings` or `schema_config`; adapt labels, properties and ID strategies +to your graph. These examples do not enumerate or migrate every graph automatically. + +The mappings support PRIMARY_KEY and CUSTOMIZE_NUMBER vertex IDs. AUTOMATIC +vertices receive new IDs, so these mappings cannot preserve their edge endpoints +losslessly. Check source ID strategies before migration. + +Every Sink mapping sets `timeZone = UTC`, assuming UTC-serialized DATE strings +from the source Server. Set each mapping to the source Server's actual time zone +when different. Source `time_zone` cannot repair timezone-free REST date strings; +include original timestamps in acceptance checks for DATE properties. + +Compare Schema types, cardinalities, labels, primary keys, endpoint definitions, +indexes and Schema IDs, plus every vertex ID/property and edge ID/endpoint/property. +Counts or job status alone are insufficient. See the official +[HugeGraph Source](https://github.com/apache/seatunnel/blob/v3.0.0/docs/en/connectors/source/HugeGraph.md) +and [Sink](https://github.com/apache/seatunnel/blob/v3.0.0/docs/en/connectors/sink/HugeGraph.md) options. + +## Limits + +Check the Workers and member counts on every hybrid Pod against the configured +replicas; Pod Ready alone does not establish a shared engine cluster. DNS +discovery's `service-port: 5801` must match the listening member port. + +- PRIMARY_KEY and CUSTOMIZE_NUMBER examples do not provide lossless AUTOMATIC-ID + copies; these require an explicit mapping of generated IDs and edge endpoints. +- The upstream 3.0.0 UI Duration display has a known timezone/completed-job issue. + Use CLI/REST status and start/end timestamps to assess actual completion. +- Stop jobs before mode changes. Uninterrupted switching, production HA and + persistent recovery are not guaranteed. Temporary checkpoints are not a durable + recovery design, and actual NetworkPolicy enforcement depends on the CNI. diff --git a/helm/hugegraph/examples/seatunnel/clone/company.conf b/helm/hugegraph/examples/seatunnel/clone/company.conf new file mode 100644 index 0000000000..aaf4b66d45 --- /dev/null +++ b/helm/hugegraph/examples/seatunnel/clone/company.conf @@ -0,0 +1,55 @@ +# 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. + +# Existing graph A -> empty graph B; copy exact Schema first through HugeGraph API. +env { job.mode = BATCH, parallelism = 1 } +source { + HugeGraph { + host = ${HG_SOURCE_HOST} + port = ${HG_SOURCE_PORT} + protocol = ${HG_SOURCE_PROTOCOL} + graph_space = DEFAULT + graph_name = graph_a + username = ${HG_SOURCE_USERNAME} + password = ${HG_SOURCE_PASSWORD} + label = "company" + label_type = VERTEX + page_size = 100 + time_zone = UTC + } +} +sink { + HugeGraph { + host = ${HG_TARGET_HOST} + port = ${HG_TARGET_PORT} + protocol = ${HG_TARGET_PROTOCOL} + graph_space = DEFAULT + graph_name = graph_b + username = ${HG_TARGET_USERNAME} + password = ${HG_TARGET_PASSWORD} + schema_save_mode = ERROR_WHEN_SCHEMA_NOT_EXIST + check_vertex = true + batch_size = 100 + mappings = [ + { + type = VERTEX + label = company + timeZone = UTC + idStrategy = CUSTOMIZE_NUMBER + idFields = ["~id"] + } + ] + } +} diff --git a/helm/hugegraph/examples/seatunnel/clone/copy-schema.py b/helm/hugegraph/examples/seatunnel/clone/copy-schema.py new file mode 100755 index 0000000000..79ace9e43a --- /dev/null +++ b/helm/hugegraph/examples/seatunnel/clone/copy-schema.py @@ -0,0 +1,108 @@ +#!/usr/bin/env python3 +# 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. + +"""Copy ordinary-label schema to an empty graph, preserving IDs for data copy.""" +import argparse +import base64 +import copy +import json +import os +import urllib.request + +KINDS = ('propertykeys', 'vertexlabels', 'edgelabels', 'indexlabels') + + +def client(base, role): + user = os.environ.get('HG_' + role + '_USERNAME', 'admin') + password = os.environ['HG_' + role + '_PASSWORD'] + token = base64.b64encode((user + ':' + password).encode()).decode() + + def request(method, suffix, body=None): + data = None if body is None else json.dumps(body).encode() + req = urllib.request.Request(base.rstrip('/') + suffix, data=data, + method=method, headers={ + 'Authorization': 'Basic ' + token, + 'Content-Type': 'application/json'}) + with urllib.request.urlopen(req, timeout=60) as response: + raw = response.read().decode() + return json.loads(raw) if raw else None + return request + + +def normalize(schema): + result = copy.deepcopy(schema) + for kind in KINDS: + for element in result[kind]: + element.pop('status', None) + element.get('user_data', {}).pop('~create_time', None) + for key in ('properties', 'nullable_keys', 'index_labels'): + if key in element: + element[key].sort() + result[kind].sort(key=lambda element: element['name']) + return result + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument('--source', required=True, help='Source graph URL, including graph name') + parser.add_argument('--target', required=True, help='Target graph URL, including graph name') + args = parser.parse_args() + source = client(args.source, 'SOURCE') + schema = source('GET', '/schema') + for label in schema['edgelabels']: + label_type = label.get('edgelabel_type', 'NORMAL') + if label_type != 'NORMAL': + raise SystemExit( + f"Unsupported edge label type {label_type!r} for {label['name']!r}: " + 'this example only supports NORMAL edge labels; ' + 'inherited PARENT/SUB edge labels are not supported.') + target = client(args.target, 'TARGET') + target_schema = target('GET', '/schema') + if any(target_schema[kind] for kind in KINDS): + raise SystemExit('Target schema must be empty; use a new graph.') + original_mode = target('GET', '/mode')['mode'] + if str(original_mode).upper() != 'NONE': + raise SystemExit('Target graph must be in NONE mode before copying schema.') + target('PUT', '/mode', 'RESTORING') + try: + for kind in KINDS: + for element in schema[kind]: + body = copy.deepcopy(element) + body.pop('status', None) + body.pop('index_labels', None) + if kind == 'indexlabels': + body['rebuild'] = False # Empty graph; subsequent data writes build indexes. + target('POST', '/schema/' + kind, body) + if normalize(schema) != normalize(target('GET', '/schema')): + raise RuntimeError('Schema comparison failed') + except Exception as error: + raise SystemExit( + 'Schema copy failed; the target remains in RESTORING mode. ' + 'Inspect it or delete and recreate the graph before copying data: ' + f'{error}') from error + try: + target('PUT', '/mode', original_mode) + except Exception as error: + raise SystemExit( + 'Schema copy verified, but restoring the target mode failed; ' + 'the target remains in RESTORING mode: ' + f'{error}') from error + print(json.dumps({'schema_equal': True, 'ids_preserved': True, + 'counts': {kind: len(schema[kind]) for kind in KINDS}})) + + +if __name__ == '__main__': + main() diff --git a/helm/hugegraph/examples/seatunnel/clone/knows.conf b/helm/hugegraph/examples/seatunnel/clone/knows.conf new file mode 100644 index 0000000000..af3960743b --- /dev/null +++ b/helm/hugegraph/examples/seatunnel/clone/knows.conf @@ -0,0 +1,57 @@ +# 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. + +# Existing graph A -> empty graph B; copy exact Schema first through HugeGraph API. +env { job.mode = BATCH, parallelism = 1 } +source { + HugeGraph { + host = ${HG_SOURCE_HOST} + port = ${HG_SOURCE_PORT} + protocol = ${HG_SOURCE_PROTOCOL} + graph_space = DEFAULT + graph_name = graph_a + username = ${HG_SOURCE_USERNAME} + password = ${HG_SOURCE_PASSWORD} + label = "knows" + label_type = EDGE + page_size = 100 + time_zone = UTC + } +} +sink { + HugeGraph { + host = ${HG_TARGET_HOST} + port = ${HG_TARGET_PORT} + protocol = ${HG_TARGET_PROTOCOL} + graph_space = DEFAULT + graph_name = graph_b + username = ${HG_TARGET_USERNAME} + password = ${HG_TARGET_PASSWORD} + schema_save_mode = ERROR_WHEN_SCHEMA_NOT_EXIST + check_vertex = true + batch_size = 100 + mappings = [ + { + type = EDGE + label = knows + timeZone = UTC + frequency = MULTIPLE + sortKeys = [since] + sourceConfig { label = person, idFields = ["~source_id"] } + targetConfig { label = person, idFields = ["~target_id"] } + } + ] + } +} diff --git a/helm/hugegraph/examples/seatunnel/clone/person.conf b/helm/hugegraph/examples/seatunnel/clone/person.conf new file mode 100644 index 0000000000..945739ad8a --- /dev/null +++ b/helm/hugegraph/examples/seatunnel/clone/person.conf @@ -0,0 +1,55 @@ +# 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. + +# Existing graph A -> empty graph B; copy exact Schema first through HugeGraph API. +env { job.mode = BATCH, parallelism = 1 } +source { + HugeGraph { + host = ${HG_SOURCE_HOST} + port = ${HG_SOURCE_PORT} + protocol = ${HG_SOURCE_PROTOCOL} + graph_space = DEFAULT + graph_name = graph_a + username = ${HG_SOURCE_USERNAME} + password = ${HG_SOURCE_PASSWORD} + label = "person" + label_type = VERTEX + page_size = 100 + time_zone = UTC + } +} +sink { + HugeGraph { + host = ${HG_TARGET_HOST} + port = ${HG_TARGET_PORT} + protocol = ${HG_TARGET_PROTOCOL} + graph_space = DEFAULT + graph_name = graph_b + username = ${HG_TARGET_USERNAME} + password = ${HG_TARGET_PASSWORD} + schema_save_mode = ERROR_WHEN_SCHEMA_NOT_EXIST + check_vertex = true + batch_size = 100 + mappings = [ + { + type = VERTEX + label = person + timeZone = UTC + idStrategy = PRIMARY_KEY + idFields = [name] + } + ] + } +} diff --git a/helm/hugegraph/examples/seatunnel/clone/works_at.conf b/helm/hugegraph/examples/seatunnel/clone/works_at.conf new file mode 100644 index 0000000000..b8a4d5f937 --- /dev/null +++ b/helm/hugegraph/examples/seatunnel/clone/works_at.conf @@ -0,0 +1,56 @@ +# 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. + +# Existing graph A -> empty graph B; copy exact Schema first through HugeGraph API. +env { job.mode = BATCH, parallelism = 1 } +source { + HugeGraph { + host = ${HG_SOURCE_HOST} + port = ${HG_SOURCE_PORT} + protocol = ${HG_SOURCE_PROTOCOL} + graph_space = DEFAULT + graph_name = graph_a + username = ${HG_SOURCE_USERNAME} + password = ${HG_SOURCE_PASSWORD} + label = "works_at" + label_type = EDGE + page_size = 100 + time_zone = UTC + } +} +sink { + HugeGraph { + host = ${HG_TARGET_HOST} + port = ${HG_TARGET_PORT} + protocol = ${HG_TARGET_PROTOCOL} + graph_space = DEFAULT + graph_name = graph_b + username = ${HG_TARGET_USERNAME} + password = ${HG_TARGET_PASSWORD} + schema_save_mode = ERROR_WHEN_SCHEMA_NOT_EXIST + check_vertex = true + batch_size = 100 + mappings = [ + { + type = EDGE + label = works_at + timeZone = UTC + frequency = SINGLE + sourceConfig { label = person, idFields = ["~source_id"] } + targetConfig { label = company, idFields = ["~target_id"] } + } + ] + } +} diff --git a/helm/hugegraph/examples/seatunnel/hugegraph-sink.conf b/helm/hugegraph/examples/seatunnel/hugegraph-sink.conf new file mode 100644 index 0000000000..7b408a2c14 --- /dev/null +++ b/helm/hugegraph/examples/seatunnel/hugegraph-sink.conf @@ -0,0 +1,67 @@ +# 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. + +# First create the graph selected by HG_GRAPH_SPACE/HG_GRAPH_NAME (default DEFAULT/graph_a). +# Helm configures the same-release HugeGraph address and Secret credential references. +env { + job.mode = "BATCH" + parallelism = 1 +} + +source { + FakeSource { + row.num = 10 + schema { + fields { + name = "string" + age = "int" + } + } + rows = [ + { kind = INSERT, fields = ["person_01", 21] }, + { kind = INSERT, fields = ["person_02", 22] }, + { kind = INSERT, fields = ["person_03", 23] }, + { kind = INSERT, fields = ["person_04", 24] }, + { kind = INSERT, fields = ["person_05", 25] }, + { kind = INSERT, fields = ["person_06", 26] }, + { kind = INSERT, fields = ["person_07", 27] }, + { kind = INSERT, fields = ["person_08", 28] }, + { kind = INSERT, fields = ["person_09", 29] }, + { kind = INSERT, fields = ["person_10", 30] } + ] + } +} + +sink { + HugeGraph { + host = ${HG_HOST} + port = ${HG_PORT} + protocol = ${HG_PROTOCOL} + graph_space = ${HG_GRAPH_SPACE} + graph_name = ${HG_GRAPH_NAME} + username = ${HG_USERNAME} + password = ${HG_PASSWORD} + schema_save_mode = "CREATE_SCHEMA_WHEN_NOT_EXIST" + mappings = [ + { + type = "VERTEX" + label = "seatunnel_demo_person" + idStrategy = "PRIMARY_KEY" + idFields = ["name"] + properties = ["name", "age"] + } + ] + } +} diff --git a/helm/hugegraph/examples/seatunnel/hugegraph-source.conf b/helm/hugegraph/examples/seatunnel/hugegraph-source.conf new file mode 100644 index 0000000000..e28a8fa148 --- /dev/null +++ b/helm/hugegraph/examples/seatunnel/hugegraph-source.conf @@ -0,0 +1,40 @@ +# 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. + +# Run hugegraph-sink.conf first, then read the ten example vertices from the same graph. +env { + job.mode = BATCH + parallelism = 1 +} + +source { + HugeGraph { + host = ${HG_HOST} + port = ${HG_PORT} + protocol = ${HG_PROTOCOL} + graph_space = ${HG_GRAPH_SPACE} + graph_name = ${HG_GRAPH_NAME} + username = ${HG_USERNAME} + password = ${HG_PASSWORD} + label = seatunnel_demo_person + label_type = VERTEX + page_size = 100 + time_zone = UTC + } +} + +sink { + Console {} +} diff --git a/helm/hugegraph/examples/values-seatunnel-cluster.yaml b/helm/hugegraph/examples/values-seatunnel-cluster.yaml new file mode 100644 index 0000000000..25308032af --- /dev/null +++ b/helm/hugegraph/examples/values-seatunnel-cluster.yaml @@ -0,0 +1,21 @@ +# 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. + +# Use the official native hybrid StatefulSet; HTTP 8080 serves REST and the task UI. +# Choose HugeGraph topology independently; ../values-single.yaml uses 1 PD + 1 Store + 1 Server. +seatunnel: + enabled: true + mode: cluster + replicas: 2 diff --git a/helm/hugegraph/examples/values-seatunnel-standalone.yaml b/helm/hugegraph/examples/values-seatunnel-standalone.yaml new file mode 100644 index 0000000000..91a79ba511 --- /dev/null +++ b/helm/hugegraph/examples/values-seatunnel-standalone.yaml @@ -0,0 +1,20 @@ +# 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. + +# Use the official native hybrid StatefulSet; HTTP 8080 serves REST and the task UI. +# Choose HugeGraph topology independently; ../values-single.yaml uses 1 PD + 1 Store + 1 Server. +seatunnel: + enabled: true + mode: standalone diff --git a/helm/hugegraph/templates/NOTES.txt b/helm/hugegraph/templates/NOTES.txt index 10892ba8ef..3ec7e7bc8d 100644 --- a/helm/hugegraph/templates/NOTES.txt +++ b/helm/hugegraph/templates/NOTES.txt @@ -24,6 +24,9 @@ limitations under the License. {{- if $hubbleEnabled }} Hubble 1 replica (UI) {{- end }} +{{- if (.Values.seatunnel | default dict).enabled }} + SeaTunnel {{ .Values.seatunnel.mode }} mode +{{- end }} {{- $inheritedEntries := list }} {{- range $component := list "server" "hubble" }} {{- $cfg := get $.Values $component }} @@ -56,6 +59,57 @@ Store waits for a PD quorum before startup. If a Pod remains in Init, inspect its init-container logs and PD readiness; see helm/hugegraph/docs/operations.md#troubleshooting. Access: +{{- if (.Values.seatunnel | default dict).enabled }} + + SeaTunnel: +{{- $stName := include "hugegraph.seatunnel.name" . }} +{{- $stService := .Values.seatunnel.service | default dict }} + In-cluster API: http://{{ $stName }}.{{ .Release.Namespace }}.svc:8080 +{{- if eq $stService.type "NodePort" }} + HTTP NodePort: + kubectl get svc -n {{ .Release.Namespace }} {{ $stName }} -o jsonpath='{.spec.ports[?(@.name=="http")].nodePort}{"\n"}' + kubectl get nodes -o wide + UI: http://:{{ $stService.nodePort | default "" }}/#/overview + API: http://:{{ $stService.nodePort | default "" }} +{{- else if eq $stService.type "LoadBalancer" }} + LoadBalancer address: + kubectl get svc -n {{ .Release.Namespace }} {{ $stName }} -o wide + kubectl get svc -n {{ .Release.Namespace }} {{ $stName }} -o jsonpath='{.status.loadBalancer.ingress}{"\n"}' + Once assigned, open http://:8080/#/overview; the API uses the same HTTP address. +{{- else }} + HTTP Service is ClusterIP; use port forwarding below for access outside the cluster. +{{- end }} + Port forwarding works with every Service type: + kubectl port-forward -n {{ .Release.Namespace }} svc/{{ $stName }} 8081:8080 + Local UI: http://127.0.0.1:8081/#/overview + Local API: http://127.0.0.1:8081 + The SeaTunnel NetworkPolicy restricts member/CLI port 5801 to same-release peers in every mode. + Enforcement requires a NetworkPolicy-capable CNI; keep namespace labels, pod creation and exec under trusted RBAC. + The UI, static resources and REST submission API use a separate SeaTunnel HTTP Basic account. + Read credentials into local variables without printing the password in installation output: + ST_USERNAME="$(kubectl get secret -n {{ .Release.Namespace }} {{ $stName }}-auth -o jsonpath='{.data.username}' | base64 --decode)" + ST_PASSWORD="$(kubectl get secret -n {{ .Release.Namespace }} {{ $stName }}-auth -o jsonpath='{.data.password}' | base64 --decode)" + Use this account for the browser Basic login; with port forwarding active, check REST with: + curl --user "${ST_USERNAME}:${ST_PASSWORD}" http://127.0.0.1:8081/overview + These entry credentials are separate from HugeGraph connector credentials, even when HugeGraph auth is disabled. + Internal CLI/5801 has no HTTP Basic authentication. External submissions must use the protected REST API. + Online Helm upgrades retain the default credentials; rotate using seatunnel.auth.password in controlled values. + Pin a password for GitOps/template-only rendering. The auth Secret is retained after disabling or uninstalling. + Authentication, rotation and jobs: helm/hugegraph/docs/seatunnel.md +{{- if and .Values.networkPolicy.enabled (has $stService.type (list "NodePort" "LoadBalancer")) }} +{{- $stRules := get .Values.networkPolicy "seatunnel" | default dict }} +{{- if empty (get $stRules "extraIngress") }} + Full NetworkPolicy isolation is enabled; allow external HTTP callers through networkPolicy.seatunnel.extraIngress. +{{- end }} +{{- end }} + + Same-release example jobs: + Server address, port, GraphSpace and credential references are injected; the job files are already mounted. + First create the target graph (default {{ .Values.server.graphSpace }}/graph_a; explicit HG_* values override it), then submit: + kubectl exec -n {{ .Release.Namespace }} sts/{{ $stName }} -- /opt/seatunnel/bin/seatunnel.sh --master cluster --config /opt/seatunnel/config/hugegraph-sink.conf + kubectl exec -n {{ .Release.Namespace }} sts/{{ $stName }} -- /opt/seatunnel/bin/seatunnel.sh --master cluster --config /opt/seatunnel/config/hugegraph-source.conf + Helm does not create graphs or run jobs. Cross-environment copies still require HG_SOURCE_* / HG_TARGET_* values. +{{- end }} {{- range $component := list "server" "hubble" "pd" }} {{- if or (ne $component "hubble") $hubbleEnabled }} {{- $cfg := get $.Values $component }} diff --git a/helm/hugegraph/templates/_helpers.tpl b/helm/hugegraph/templates/_helpers.tpl index b59a3fb9c2..5094d48670 100644 --- a/helm/hugegraph/templates/_helpers.tpl +++ b/helm/hugegraph/templates/_helpers.tpl @@ -23,9 +23,12 @@ {{- if not (hasKey .Values "hubble") -}} {{- $_ := set .Values "hubble" (dict "enabled" false "service" dict "persistence" dict) -}} {{- end -}} +{{- if not (hasKey .Values "seatunnel") -}} +{{- $_ := set .Values "seatunnel" (dict "enabled" false "service" dict) -}} +{{- end -}} {{- $strict := eq (.Values.mode | default "easy") "strict" -}} {{- $serviceType := ternary "ClusterIP" "NodePort" (or $strict $legacy) -}} -{{- range $comp := list "server" "hubble" -}} +{{- range $comp := list "server" "hubble" "seatunnel" -}} {{- $component := get $.Values $comp -}} {{- $svc := get $component "service" | default dict -}} {{- $_ := set $component "service" $svc -}} @@ -48,7 +51,7 @@ {{- $registry := trimSuffix "/" (get (.Values.global | default dict) "imageRegistry" | default "") -}} {{- if contains "://" $registry -}}{{- fail "global.imageRegistry must be a registry host/path without http:// or https://" -}}{{- end -}} {{- if $registry -}} -{{- range $comp := list "pd" "store" "server" "hubble" -}} +{{- range $comp := list "pd" "store" "server" "hubble" "seatunnel" -}} {{- $image := get (get $.Values $comp) "image" | default dict -}} {{- if $image -}} {{- $_ := set $image "repository" (include "hugegraph.registryImage" (dict "reference" $image.repository "registry" $registry)) -}} @@ -749,7 +752,28 @@ The minimum Server replica count that a PDB must remain valid against. Cross-field validation that JSON Schema draft-07 cannot express. */}} {{- define "hugegraph.validateValues" -}} +{{- if (.Values.seatunnel | default dict).enabled -}} +{{- if not (has .Values.seatunnel.mode (list "standalone" "cluster")) -}} +{{- fail "seatunnel.mode must be standalone or cluster" -}} +{{- end -}} +{{- if and (eq .Values.seatunnel.mode "cluster") (lt (int .Values.seatunnel.replicas) 2) -}} +{{- fail "seatunnel.replicas must be at least 2 in cluster mode" -}} +{{- end -}} +{{- end -}} + {{- include "hugegraph.resolveValues" . -}} +{{- if (.Values.seatunnel | default dict).enabled -}} +{{- $svc := .Values.seatunnel.service -}} +{{- if and (get $svc "nodePort") (not (has $svc.type (list "NodePort" "LoadBalancer"))) -}} +{{- fail "seatunnel.service.nodePort requires seatunnel.service.type to be NodePort or LoadBalancer" -}} +{{- end -}} +{{- range $key := list "username" "password" -}} +{{- $value := get ($.Values.seatunnel.auth | default dict) $key | default "" -}} +{{- if or (ne $value (trim $value)) (contains "${" $value) -}} +{{- fail (printf "seatunnel.auth.%s cannot contain ${...} placeholders or leading/trailing whitespace: native Engine YAML expands or trims them" $key) -}} +{{- end -}} +{{- end -}} +{{- end -}} {{- $startupBudget := include "hugegraph.server.startupProbeBudget" . | int -}} {{- if lt $startupBudget (add (int .Values.server.storageWaitTimeoutSeconds) 120) -}} {{- fail "server.probes.startup.failureThreshold and periodSeconds must leave at least server.storageWaitTimeoutSeconds + 120 seconds to wait for storage and start Server; increase the probe budget or reduce server.storageWaitTimeoutSeconds" -}} @@ -1044,6 +1068,22 @@ resolver (CoreDNS in any namespace, NodeLocal DNSCache). {{- join "\n" $rules -}} {{- end }} +{{/* Native SeaTunnel resources belong to this HugeGraph release. */}} +{{- define "hugegraph.seatunnel.name" -}} +{{- printf "%s-seatunnel" (include "hugegraph.fullname" . | trunc 44 | trimSuffix "-") -}} +{{- end }} + +{{- define "hugegraph.netpol.seatunnelLabels" -}} +{{ include "hugegraph.selectorLabels" . }} +app.kubernetes.io/component: seatunnel +{{- end }} + +{{- define "hugegraph.netpol.seatunnelPeer" -}} +- podSelector: + matchLabels: + {{- include "hugegraph.netpol.seatunnelLabels" . | nindent 6 }} +{{- end }} + {{- define "hugegraph.netpol.dns" -}} - ports: - protocol: UDP diff --git a/helm/hugegraph/templates/networkpolicy.yaml b/helm/hugegraph/templates/networkpolicy.yaml index e8638d6d74..cbe2f4ccac 100644 --- a/helm/hugegraph/templates/networkpolicy.yaml +++ b/helm/hugegraph/templates/networkpolicy.yaml @@ -135,6 +135,9 @@ spec: {{- if $hubbleEnabled }} {{- include "hugegraph.netpol.peer" (dict "root" . "component" "hubble") | nindent 8 }} {{- end }} + {{- if (.Values.seatunnel | default dict).enabled }} + {{- include "hugegraph.netpol.seatunnelPeer" . | nindent 8 }} + {{- end }} ports: {{- include "hugegraph.netpol.ports" (list $serverPort) | nindent 8 }} {{- with get (get $np "server" | default dict) "extraIngress" }} diff --git a/helm/hugegraph/templates/seatunnel-auth-secret.yaml b/helm/hugegraph/templates/seatunnel-auth-secret.yaml new file mode 100644 index 0000000000..918f1a5ac9 --- /dev/null +++ b/helm/hugegraph/templates/seatunnel-auth-secret.yaml @@ -0,0 +1,74 @@ +{{/* +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. +*/}} + +{{- define "hugegraph.seatunnel.authSecretName" -}} +{{- printf "%s-auth" (include "hugegraph.seatunnel.name" .) -}} +{{- end -}} + +{{/* Resolve a base64 credential and reuse the live Secret during upgrades. */}} +{{- define "hugegraph.seatunnel.authPassword" -}} +{{- $auth := get .Values.seatunnel "auth" | default dict -}} +{{- $password := get $auth "password" | default "" -}} +{{- if $password -}} +{{- $password | b64enc -}} +{{- else -}} +{{- $name := include "hugegraph.seatunnel.authSecretName" . -}} +{{- $secret := lookup "v1" "Secret" .Release.Namespace $name -}} +{{- if and $secret (hasKey $secret "data") (hasKey (get $secret "data") "password") -}} +{{- get (get $secret "data") "password" -}} +{{- else -}} +{{- include "hugegraph.generatedCredential" (dict "root" . "name" "seatunnel-admin") | b64enc -}} +{{- end -}} +{{- end -}} +{{- end -}} + +{{/* Engine YAML uses JVM properties, not Pod env; mount its password in a Secret. */}} +{{- define "hugegraph.seatunnel.engineConfig" -}} +{{- $files := include "hugegraph.seatunnel.config" . | fromYaml -}} +{{- $config := get $files "seatunnel.yaml" | fromYaml | default dict -}} +{{- if hasKey $config "Error" -}} +{{- fail "seatunnel.config[seatunnel.yaml] must contain valid native YAML" -}} +{{- end -}} +{{- $seatunnel := get $config "seatunnel" | default dict -}} +{{- $engine := get $seatunnel "engine" | default dict -}} +{{- $http := get $engine "http" | default dict -}} +{{- $auth := get .Values.seatunnel "auth" | default dict -}} +{{- $_ := set $http "enable-basic-auth" true -}} +{{- $_ := set $http "basic-auth-username" (get $auth "username" | default "admin") -}} +{{- $_ := set $http "basic-auth-password" (include "hugegraph.seatunnel.authPassword" . | b64dec) -}} +{{- $_ := set $engine "http" $http -}} +{{- $_ := set $seatunnel "engine" $engine -}} +{{- $_ := set $config "seatunnel" $seatunnel -}} +{{- $config | toYaml -}} +{{- end -}} + +{{- if (.Values.seatunnel | default dict).enabled }} +apiVersion: v1 +kind: Secret +metadata: + name: {{ include "hugegraph.seatunnel.authSecretName" . }} + labels: + {{- include "hugegraph.labels" . | nindent 4 }} + app.kubernetes.io/component: seatunnel + annotations: + helm.sh/resource-policy: keep +type: Opaque +data: + username: {{ (get (get .Values.seatunnel "auth" | default dict) "username" | default "admin") | b64enc | quote }} + password: {{ include "hugegraph.seatunnel.authPassword" . | quote }} + seatunnel.yaml: {{ include "hugegraph.seatunnel.engineConfig" . | b64enc | quote }} +{{- end }} diff --git a/helm/hugegraph/templates/seatunnel-networkpolicy.yaml b/helm/hugegraph/templates/seatunnel-networkpolicy.yaml new file mode 100644 index 0000000000..faddf24c86 --- /dev/null +++ b/helm/hugegraph/templates/seatunnel-networkpolicy.yaml @@ -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. + +{{ include "hugegraph.resolveValues" . }} +{{- if (.Values.seatunnel | default dict).enabled }} +{{- $rules := get .Values.networkPolicy "seatunnel" | default dict }} +apiVersion: networking.k8s.io/v1 +kind: NetworkPolicy +metadata: + name: {{ include "hugegraph.fullname" . | trunc 53 | trimSuffix "-" }}-seatunnel + labels: + {{- include "hugegraph.labels" . | nindent 4 }} + app.kubernetes.io/component: seatunnel +spec: + podSelector: + matchLabels: + {{- include "hugegraph.netpol.seatunnelLabels" . | nindent 6 }} + policyTypes: + - Ingress + {{- if .Values.networkPolicy.enabled }} + - Egress + {{- end }} + ingress: + - from: + {{- include "hugegraph.netpol.seatunnelPeer" . | nindent 8 }} + ports: + {{- include "hugegraph.netpol.ports" (list 5801 8080) | nindent 8 }} + {{- if not .Values.networkPolicy.enabled }} + # Easy mode keeps the authenticated HTTP entry open, never the member port. + - ports: + {{- include "hugegraph.netpol.ports" (list 8080) | nindent 8 }} + {{- end }} + {{- with get $rules "extraIngress" }} + {{- toYaml . | nindent 4 }} + {{- end }} + {{- if .Values.networkPolicy.enabled }} + egress: + - to: + {{- include "hugegraph.netpol.seatunnelPeer" . | nindent 8 }} + ports: + {{- include "hugegraph.netpol.ports" (list 5801 8080) | nindent 8 }} + - to: + {{- include "hugegraph.netpol.peer" (dict "root" . "component" "server") | nindent 8 }} + ports: + {{- include "hugegraph.netpol.ports" (list .Values.server.port) | nindent 8 }} + {{- include "hugegraph.netpol.dns" . | nindent 4 }} + {{- with get $rules "extraEgress" }} + {{- toYaml . | nindent 4 }} + {{- end }} + {{- end }} +{{- end }} diff --git a/helm/hugegraph/templates/seatunnel.yaml b/helm/hugegraph/templates/seatunnel.yaml new file mode 100644 index 0000000000..fabe1f269e --- /dev/null +++ b/helm/hugegraph/templates/seatunnel.yaml @@ -0,0 +1,262 @@ +{{/* +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. + +Based on Apache SeaTunnel v3.0.0 hybrid Kubernetes resources: +https://github.com/apache/seatunnel/blob/v3.0.0/docs/zh/getting-started/kubernetes/hybrid-cluster-mode.md +Use its DNS discovery alternative, so Kubernetes API RBAC is unnecessary. +*/}} + +{{- define "hugegraph.seatunnel.config" -}} +{{- $name := include "hugegraph.seatunnel.name" . -}} +{{- $dns := printf "%s-headless.%s.svc" $name .Release.Namespace -}} +{{- $data := dict -}} +{{- $hazelcast := dict "hazelcast" (dict + "cluster-name" $name + "network" (dict + "port" (dict "auto-increment" false "port" 5801) + "join" (dict "kubernetes" (dict "enabled" true "service-dns" $dns "service-port" 5801 "service-dns-timeout" 10))) + "properties" (dict "hazelcast.logging.type" "log4j2")) -}} +{{- $_ := set $data "hazelcast.yaml" ($hazelcast | toYaml) -}} +{{- $client := dict "hazelcast-client" (dict + "cluster-name" $name + "properties" (dict "hazelcast.logging.type" "log4j2") + "connection-strategy" (dict "connection-retry" (dict "cluster-connect-timeout-millis" 7000)) + "network" (dict "cluster-members" (list (printf "%s:5801" $dns)))) -}} +{{- $_ := set $data "hazelcast-client.yaml" ($client | toYaml) -}} +{{- $engine := dict "seatunnel" (dict "engine" (dict + "backup-count" (ternary 0 1 (eq .Values.seatunnel.mode "standalone")) + "history-job-expire-minutes" 1440 + "classloader-cache-mode" true + "slot-service" (dict "dynamic-slot" true) + "checkpoint" (dict "interval" 180000 "timeout" 30000 "storage" (dict + "type" "hdfs" "max-retained" 3 "plugin-config" (dict + "namespace" "/tmp/seatunnel/checkpoint/" "storage.type" "hdfs" "fs.defaultFS" "file:///tmp/"))) + "http" (dict "enable-http" true "port" 8080 "enable-dynamic-port" false))) -}} +{{- $_ := set $data "seatunnel.yaml" ($engine | toYaml) -}} +{{/* The upstream script appends jvm_options after JAVA_OPTS; set heap in that file. */}} +{{- $_ := set $data "jvm_options" (.Values.seatunnel.javaOpts | replace " " "\n") -}} +{{- range $path, $_ := .Files.Glob "examples/seatunnel/hugegraph-*.conf" -}} +{{- $_ := set $data (base $path) ($.Files.Get $path) -}} +{{- end -}} +{{- range $filename, $content := .Values.seatunnel.config -}} +{{- $_ := set $data $filename $content -}} +{{- end -}} +{{- toYaml $data -}} +{{- end -}} + +{{- if (.Values.seatunnel | default dict).enabled }} +{{- include "hugegraph.resolveValues" . }} +{{- include "hugegraph.validateValues" . }} +{{- $name := include "hugegraph.seatunnel.name" . }} +{{- $config := omit (include "hugegraph.seatunnel.config" . | fromYaml) "seatunnel.yaml" }} +{{- $svc := .Values.seatunnel.service }} +{{- $env := list + (dict "name" "SEATUNNEL_HOME" "value" "/opt/seatunnel") + (dict "name" "HG_HOST" "value" (printf "%s.%s.svc" (include "hugegraph.server.name" .) .Release.Namespace)) + (dict "name" "HG_PORT" "value" (.Values.server.port | toString)) + (dict "name" "HG_PROTOCOL" "value" "http") + (dict "name" "HG_USERNAME" "value" "admin") + (dict "name" "HG_GRAPH_SPACE" "value" .Values.server.graphSpace) + (dict "name" "HG_GRAPH_NAME" "value" "graph_a") }} +{{- $password := dict "name" "HG_PASSWORD" "value" "" }} +{{- if .Values.server.auth.enabled }} +{{- $password = dict "name" "HG_PASSWORD" "valueFrom" (dict "secretKeyRef" (dict + "name" (include "hugegraph.server.authSecretName" .) + "key" (include "hugegraph.server.authSecretKey" .))) }} +{{- end }} +{{- $env = append $env $password }} +{{- $overrides := dict }} +{{- range .Values.seatunnel.env }} +{{- $_ := set $overrides .name true }} +{{- end }} +apiVersion: v1 +kind: ConfigMap +metadata: + name: {{ $name }}-configs + labels: + {{- include "hugegraph.labels" . | nindent 4 }} + app.kubernetes.io/component: seatunnel +data: + {{- toYaml $config | nindent 2 }} +--- +apiVersion: v1 +kind: Service +metadata: + name: {{ $name }}-headless + labels: + {{- include "hugegraph.labels" . | nindent 4 }} + app.kubernetes.io/component: seatunnel +spec: + clusterIP: None + publishNotReadyAddresses: true + ports: + - name: hazelcast + port: 5801 + targetPort: hazelcast + selector: + {{- include "hugegraph.selectorLabels" . | nindent 4 }} + app.kubernetes.io/component: seatunnel +--- +apiVersion: v1 +kind: Service +metadata: + name: {{ $name }} + labels: + {{- include "hugegraph.labels" . | nindent 4 }} + app.kubernetes.io/component: seatunnel + {{- with $svc.annotations }} + annotations: + {{- toYaml . | nindent 4 }} + {{- end }} +spec: + type: {{ $svc.type }} + {{- range $field := list "externalTrafficPolicy" "loadBalancerClass" "loadBalancerIP" "loadBalancerSourceRanges" "externalIPs" }} + {{- with get $svc $field }} + {{ $field }}: {{ toYaml . | nindent 4 }} + {{- end }} + {{- end }} + {{- if hasKey $svc "allocateLoadBalancerNodePorts" }} + {{- if ne (get $svc "allocateLoadBalancerNodePorts") nil }} + allocateLoadBalancerNodePorts: {{ $svc.allocateLoadBalancerNodePorts }} + {{- end }} + {{- end }} + ports: + - name: http + port: 8080 + targetPort: http + {{- with $svc.nodePort }} + nodePort: {{ . }} + {{- end }} + selector: + {{- include "hugegraph.selectorLabels" . | nindent 4 }} + app.kubernetes.io/component: seatunnel +--- +apiVersion: apps/v1 +kind: StatefulSet +metadata: + name: {{ $name }} + labels: + {{- include "hugegraph.labels" . | nindent 4 }} + app.kubernetes.io/component: seatunnel +spec: + serviceName: {{ $name }}-headless + replicas: {{ ternary 1 (int .Values.seatunnel.replicas) (eq .Values.seatunnel.mode "standalone") }} + podManagementPolicy: Parallel + selector: + matchLabels: + {{- include "hugegraph.selectorLabels" . | nindent 6 }} + app.kubernetes.io/component: seatunnel + template: + metadata: + labels: + {{- include "hugegraph.selectorLabels" . | nindent 8 }} + app.kubernetes.io/component: seatunnel + annotations: + checksum/config: {{ $config | toYaml | sha256sum }} + checksum/auth: {{ include "hugegraph.seatunnel.engineConfig" . | sha256sum }} + spec: + automountServiceAccountToken: false + terminationGracePeriodSeconds: 120 + {{- with .Values.imagePullSecrets }} + imagePullSecrets: + {{- toYaml . | nindent 8 }} + {{- end }} + {{- with .Values.seatunnel.priorityClassName }} + priorityClassName: {{ . | quote }} + {{- end }} + {{- with .Values.seatunnel.nodeSelector }} + nodeSelector: + {{- toYaml . | nindent 8 }} + {{- end }} + {{- with .Values.seatunnel.tolerations }} + tolerations: + {{- toYaml . | nindent 8 }} + {{- end }} + {{- with .Values.seatunnel.affinity }} + affinity: + {{- toYaml . | nindent 8 }} + {{- end }} + {{- with .Values.seatunnel.podSecurityContext }} + securityContext: + {{- toYaml . | nindent 8 }} + {{- end }} + containers: + - name: seatunnel + image: {{ include "hugegraph.image" (dict "image" .Values.seatunnel.image "appVersion" "3.0.0") | quote }} + imagePullPolicy: {{ .Values.seatunnel.image.pullPolicy }} + {{- with .Values.seatunnel.securityContext }} + securityContext: + {{- toYaml . | nindent 12 }} + {{- end }} + # The default upstream role is hybrid; an explicit -r is unnecessary. + command: ["/opt/seatunnel/bin/seatunnel-cluster.sh"] + env: + {{- range $env }} + {{- if not (hasKey $overrides .name) }} + {{- toYaml (list .) | nindent 12 }} + {{- end }} + {{- end }} + {{- with .Values.seatunnel.env }} + {{- toYaml . | nindent 12 }} + {{- end }} + ports: + - name: http + containerPort: 8080 + - name: hazelcast + containerPort: 5801 + {{- with .Values.seatunnel.resources }} + resources: + {{- toYaml . | nindent 12 }} + {{- end }} + startupProbe: + tcpSocket: + port: http + periodSeconds: 5 + failureThreshold: 60 + readinessProbe: + tcpSocket: + port: http + periodSeconds: 10 + timeoutSeconds: 5 + livenessProbe: + tcpSocket: + port: hazelcast + periodSeconds: 30 + timeoutSeconds: 5 + lifecycle: + preStop: + exec: + command: ["/opt/seatunnel/bin/stop-seatunnel-cluster.sh"] + volumeMounts: + - name: seatunnel-auth + mountPath: /opt/seatunnel/config/seatunnel.yaml + subPath: seatunnel.yaml + {{- range $filename, $_ := $config }} + - name: seatunnel-configs + mountPath: /opt/seatunnel/config/{{ $filename }} + subPath: {{ $filename }} + {{- end }} + volumes: + - name: seatunnel-auth + secret: + secretName: {{ include "hugegraph.seatunnel.authSecretName" . }} + items: + - key: seatunnel.yaml + path: seatunnel.yaml + - name: seatunnel-configs + configMap: + name: {{ $name }}-configs +{{- end }} diff --git a/helm/hugegraph/tests/__snapshot__/networkpolicy_test.yaml.snap b/helm/hugegraph/tests/__snapshot__/networkpolicy_test.yaml.snap index a79353de84..a836601261 100644 --- a/helm/hugegraph/tests/__snapshot__/networkpolicy_test.yaml.snap +++ b/helm/hugegraph/tests/__snapshot__/networkpolicy_test.yaml.snap @@ -26,7 +26,7 @@ matches the reviewed cluster plus Hubble render: app.kubernetes.io/managed-by: Helm app.kubernetes.io/name: hugegraph app.kubernetes.io/version: latest - helm.sh/chart: hugegraph-helm-0.2.0 + helm.sh/chart: hugegraph-helm-0.3.0 name: RELEASE-NAME-hugegraph-pd spec: egress: @@ -97,7 +97,7 @@ matches the reviewed cluster plus Hubble render: app.kubernetes.io/managed-by: Helm app.kubernetes.io/name: hugegraph app.kubernetes.io/version: latest - helm.sh/chart: hugegraph-helm-0.2.0 + helm.sh/chart: hugegraph-helm-0.3.0 name: RELEASE-NAME-hugegraph-store spec: egress: @@ -174,7 +174,7 @@ matches the reviewed cluster plus Hubble render: app.kubernetes.io/managed-by: Helm app.kubernetes.io/name: hugegraph app.kubernetes.io/version: latest - helm.sh/chart: hugegraph-helm-0.2.0 + helm.sh/chart: hugegraph-helm-0.3.0 name: RELEASE-NAME-hugegraph-server spec: egress: @@ -238,7 +238,7 @@ matches the reviewed cluster plus Hubble render: app.kubernetes.io/managed-by: Helm app.kubernetes.io/name: hugegraph app.kubernetes.io/version: latest - helm.sh/chart: hugegraph-helm-0.2.0 + helm.sh/chart: hugegraph-helm-0.3.0 name: RELEASE-NAME-hugegraph-hubble spec: egress: diff --git a/helm/hugegraph/tests/networkpolicy_test.yaml b/helm/hugegraph/tests/networkpolicy_test.yaml index edc7f6e835..91124ceb86 100644 --- a/helm/hugegraph/tests/networkpolicy_test.yaml +++ b/helm/hugegraph/tests/networkpolicy_test.yaml @@ -832,7 +832,7 @@ tests: documentIndex: 0 - equal: path: metadata.labels["helm.sh/chart"] - value: hugegraph-helm-0.2.0 + value: hugegraph-helm-0.3.0 documentIndex: 0 - equal: path: metadata.labels["app.kubernetes.io/version"] @@ -840,7 +840,7 @@ tests: documentIndex: 1 - equal: path: metadata.labels["helm.sh/chart"] - value: hugegraph-helm-0.2.0 + value: hugegraph-helm-0.3.0 documentIndex: 1 - equal: path: metadata.labels["app.kubernetes.io/version"] @@ -848,7 +848,7 @@ tests: documentIndex: 2 - equal: path: metadata.labels["helm.sh/chart"] - value: hugegraph-helm-0.2.0 + value: hugegraph-helm-0.3.0 documentIndex: 2 - equal: path: metadata.labels["app.kubernetes.io/version"] @@ -856,7 +856,7 @@ tests: documentIndex: 3 - equal: path: metadata.labels["helm.sh/chart"] - value: hugegraph-helm-0.2.0 + value: hugegraph-helm-0.3.0 documentIndex: 3 - it: appends extra rules verbatim diff --git a/helm/hugegraph/values.schema.json b/helm/hugegraph/values.schema.json index 987e650c16..4c0713c864 100644 --- a/helm/hugegraph/values.schema.json +++ b/helm/hugegraph/values.schema.json @@ -58,6 +58,61 @@ "type": "boolean" } } + }, + "seatunnel": { + "type": "object", + "description": "Optional native SeaTunnel hybrid Kubernetes deployment.", + "properties": { + "enabled": { + "type": "boolean" + }, + "mode": { + "type": "string", + "enum": [ + "standalone", + "cluster" + ] + }, + "replicas": { + "type": "integer", + "minimum": 1 + }, + "service": { + "$ref": "#/definitions/service" + }, + "auth": { + "type": "object", + "additionalProperties": false, + "properties": { + "username": { + "type": "string", + "minLength": 1, + "pattern": "^[^:]+$" + }, + "password": { + "type": "string" + } + } + }, + "podSecurityContext": { + "type": "object" + }, + "securityContext": { + "type": "object" + }, + "nodeSelector": { + "type": "object" + }, + "tolerations": { + "type": "array" + }, + "affinity": { + "type": "object" + }, + "priorityClassName": { + "type": "string" + } + } } }, "definitions": { @@ -128,6 +183,18 @@ "$ref": "#/definitions/networkPolicyEgressRules" } } + }, + "seatunnel": { + "type": "object", + "additionalProperties": false, + "properties": { + "extraIngress": { + "$ref": "#/definitions/networkPolicyIngressRules" + }, + "extraEgress": { + "$ref": "#/definitions/networkPolicyEgressRules" + } + } } } }, diff --git a/helm/hugegraph/values.yaml b/helm/hugegraph/values.yaml index 8ca4a2598a..0d49297668 100644 --- a/helm/hugegraph/values.yaml +++ b/helm/hugegraph/values.yaml @@ -542,9 +542,64 @@ hubble: failureThreshold: 3 timeoutSeconds: 5 +# Optional engine using Apache SeaTunnel's native hybrid Kubernetes deployment. +# One hybrid StatefulSet serves both modes; HTTP includes the built-in Web UI. +seatunnel: + enabled: false + mode: standalone + # Used in cluster mode; standalone always starts one hybrid node. + replicas: 2 + image: + repository: apache/seatunnel + tag: "3.0.0" + pullPolicy: IfNotPresent + service: + # Unset inherits mode (easy: NodePort, strict: ClusterIP). + type: + # Omitted/0 allocates automatically; only the HTTP/UI port is exposed. + nodePort: + annotations: {} + auth: + # Native HTTP Basic authentication for the Web UI and all REST endpoints. + # Independent of the HugeGraph credential used by source/sink connectors. + username: admin + # Empty generates a kept Secret; upgrades reuse the existing password. + password: "" + # Replaces the image's jvm_options; reserve memory outside the heap under 2Gi. + javaOpts: >- + -Xms512m -Xmx1g -XX:+UseG1GC -XX:MaxMetaspaceSize=256m + -XX:+HeapDumpOnOutOfMemoryError -XX:HeapDumpPath=/tmp/seatunnel.hprof + resources: + requests: + cpu: 500m + memory: 1536Mi + limits: + memory: 2Gi + # Match the other workloads. The upstream image runs as root; configure a + # compatible UID and writable paths before enabling runAsNonRoot. + podSecurityContext: {} + securityContext: + allowPrivilegeEscalation: false + capabilities: + drop: ["ALL"] + seccompProfile: + type: RuntimeDefault + nodeSelector: {} + tolerations: [] + affinity: {} + priorityClassName: "" + # Native Kubernetes env entries; matching names override automatic HG_* values. + # HG_HOST/PORT/PROTOCOL, HG_USERNAME/PASSWORD and HG_GRAPH_SPACE/NAME default to + # this release's Server, existing admin Secret, graph space and graph_a. + env: [] + # Full native configuration files, keyed by filename (e.g. seatunnel.yaml). + # Keep member port 5801 and HTTP 8080 when overriding discovery/engine files. + config: {} + # Kubernetes NetworkPolicy: mode default plus explicit per-component rules. -# easy does not isolate clients; strict isolates release traffic and DNS. Only -# enforced when the network plugin implements NetworkPolicy (kind v0.25+, +# easy keeps HTTP clients open; strict isolates release traffic and DNS. +# Enabled SeaTunnel always restricts member port 5801 to its release peers. +# Enforced only when the network plugin implements NetworkPolicy (kind v0.25+, # k3s, Calico, Cilium); elsewhere the objects are accepted and do nothing. # Nothing outside the release is admitted unless it is listed in # .extraIngress as standard NetworkPolicy ingress rules: apps in @@ -565,6 +620,13 @@ networkPolicy: # For Hubble's optional outside endpoints (es.urls, prometheus.url set # through extraEnv). PD, Store and Server reach only the release and DNS. extraEgress: [] + seatunnel: + # Member port 5801 is restricted even when networkPolicy.enabled is false. + # With full isolation, peers, DNS and the Server API are allowed by default. + # Add trusted submitters/monitoring using native NetworkPolicy rules. + extraIngress: [] + # External source/sink systems and shared checkpoint storage need egress. + extraEgress: [] # No init Job; see README.md for the HStore initialization contract. # Install with: helm install ... --wait (no --wait-for-jobs) diff --git a/helm/schema_copy/__init__.py b/helm/schema_copy/__init__.py new file mode 100644 index 0000000000..a6eb1b544f --- /dev/null +++ b/helm/schema_copy/__init__.py @@ -0,0 +1,14 @@ +# 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. diff --git a/helm/schema_copy/test_helm_seatunnel.py b/helm/schema_copy/test_helm_seatunnel.py new file mode 100755 index 0000000000..6db464873a --- /dev/null +++ b/helm/schema_copy/test_helm_seatunnel.py @@ -0,0 +1,121 @@ +#!/usr/bin/env python3 +# 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. + +"""Failure-safety regression tests for the SeaTunnel schema-copy helper.""" + +import contextlib +import copy +import importlib.util +import io +from pathlib import Path +import sys +import unittest +from unittest import mock + + +SCRIPT = (Path(__file__).resolve().parents[1] / + "hugegraph/examples/seatunnel/clone/copy-schema.py") +SPEC = importlib.util.spec_from_file_location("seatunnel_copy_schema", SCRIPT) +COPY_SCHEMA = importlib.util.module_from_spec(SPEC) +SPEC.loader.exec_module(COPY_SCHEMA) + + +def schema(property_names=()): + result = {kind: [] for kind in COPY_SCHEMA.KINDS} + result["propertykeys"] = [ + {"id": index + 1, "name": name, "data_type": "TEXT", + "cardinality": "SINGLE", "user_data": {}} + for index, name in enumerate(property_names) + ] + return result + + +class SchemaCopyFailureSafetyTest(unittest.TestCase): + + def run_copy(self, source_schema, target): + source = mock.Mock(return_value=source_schema) + with mock.patch.object(COPY_SCHEMA, "client", + side_effect=[source, target]), \ + mock.patch.object(sys, "argv", [str(SCRIPT), + "--source", "http://source/graph", + "--target", "http://target/graph"]): + return COPY_SCHEMA.main() + + def test_partial_post_failure_keeps_target_in_restoring_mode(self): + source_schema = schema(("name", "city")) + calls = [] + posts = 0 + + def target(method, suffix, body=None): + nonlocal posts + calls.append((method, suffix, body)) + if (method, suffix) == ("GET", "/schema"): + return schema() + if (method, suffix) == ("GET", "/mode"): + return {"mode": "NONE"} + if method == "POST": + posts += 1 + if posts == 2: + raise RuntimeError("injected schema write failure") + return None + + with self.assertRaisesRegex(SystemExit, "remains in RESTORING mode"): + self.run_copy(source_schema, target) + self.assertEqual(["RESTORING"], [body for method, suffix, body in calls + if (method, suffix) == ("PUT", "/mode")]) + + def test_schema_mismatch_keeps_target_in_restoring_mode(self): + source_schema = schema(("name",)) + calls = [] + schema_reads = iter((schema(), schema())) + + def target(method, suffix, body=None): + calls.append((method, suffix, body)) + if (method, suffix) == ("GET", "/schema"): + return next(schema_reads) + if (method, suffix) == ("GET", "/mode"): + return {"mode": "NONE"} + return None + + with self.assertRaisesRegex(SystemExit, "Schema comparison failed"): + self.run_copy(source_schema, target) + self.assertEqual(["RESTORING"], [body for method, suffix, body in calls + if (method, suffix) == ("PUT", "/mode")]) + + def test_verified_copy_restores_original_mode(self): + source_schema = schema(("name",)) + calls = [] + schema_reads = iter((schema(), copy.deepcopy(source_schema))) + + def target(method, suffix, body=None): + calls.append((method, suffix, body)) + if (method, suffix) == ("GET", "/schema"): + return next(schema_reads) + if (method, suffix) == ("GET", "/mode"): + return {"mode": "NONE"} + return None + + output = io.StringIO() + with contextlib.redirect_stdout(output): + self.run_copy(source_schema, target) + self.assertEqual(["RESTORING", "NONE"], + [body for method, suffix, body in calls + if (method, suffix) == ("PUT", "/mode")]) + self.assertIn('"schema_equal": true', output.getvalue()) + + +if __name__ == "__main__": + unittest.main() diff --git a/helm/test_helm_seatunnel.py b/helm/test_helm_seatunnel.py new file mode 100644 index 0000000000..5f5e62e139 --- /dev/null +++ b/helm/test_helm_seatunnel.py @@ -0,0 +1,476 @@ +#!/usr/bin/env python3 +# 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. + +"""Check native hybrid deployment, release isolation, connectivity and packaging.""" + +import base64 +import importlib.util +from pathlib import Path +import shutil +import subprocess +import tarfile +import tempfile +import unittest +from unittest import mock + +import yaml + + +CHART = Path(__file__).resolve().parent / "hugegraph" +HELM_TEMPLATE_TIMEOUT_SECONDS = 120 +HELM_PACKAGE_TIMEOUT_SECONDS = 300 + + +def matches(selector, labels): + return all(labels.get(key) == value for key, value in selector.items()) + + +def engine_resources(documents, kind=None): + return [doc for doc in documents if (kind is None or doc["kind"] == kind) + and doc["metadata"].get("labels", {}).get("app.kubernetes.io/component") == "seatunnel"] + + +def pod(workload): + return workload["spec"]["template"] + + +def engine_auth_secret(documents): + return next(doc for doc in engine_resources(documents, "Secret") + if "seatunnel.yaml" in doc.get("data", {})) + + +def engine_config(documents): + data = {} + for doc in engine_resources(documents, "ConfigMap"): + data.update(doc["data"]) + data["seatunnel.yaml"] = base64.b64decode(engine_auth_secret(documents)["data"]["seatunnel.yaml"]).decode() + return data + + +class SeaTunnelChartTest(unittest.TestCase): + @classmethod + def setUpClass(cls): + if shutil.which("helm") is None: + raise RuntimeError("Helm is required for the SeaTunnel integration regression") + + def run_helm(self, arguments, timeout, check=False): + command = ["helm", *arguments] + try: + return subprocess.run(command, check=check, capture_output=True, + text=True, timeout=timeout) + except subprocess.TimeoutExpired as error: + self.fail(f"{' '.join(command)} timed out after {timeout} seconds: {error}") + + def render(self, seatunnel=None, chart=CHART, namespace="default", release="fixture", **values): + if seatunnel is not None: + values["seatunnel"] = {"enabled": True, **seatunnel} + # Keep independent render comparisons deterministic; generation is tested separately. + values["seatunnel"]["auth"] = {"password": "native-test-admin-password", + **seatunnel.get("auth", {})} + with tempfile.TemporaryDirectory(prefix="seatunnel-values-") as directory: + configuration = Path(directory) / "values.yaml" + configuration.write_text(yaml.safe_dump(values)) + result = self.run_helm(["template", release, str(chart), + "--namespace", namespace, "-f", str(configuration)], + HELM_TEMPLATE_TIMEOUT_SECONDS) + return result + + def documents(self, seatunnel=None, chart=CHART, namespace="default", release="fixture", **values): + result = self.render(seatunnel, chart=chart, namespace=namespace, release=release, **values) + self.assertEqual(0, result.returncode, result.stderr) + return [doc for doc in yaml.safe_load_all(result.stdout) if doc] + + def test_disabled_has_no_engine_resources(self): + documents = self.documents() + self.assertEqual([], engine_resources(documents)) + self.assertFalse(any("seatunnel" in doc["metadata"]["name"] for doc in documents)) + + def test_schema_copy_rejects_unsupported_edges_before_contacting_target(self): + path = CHART / "examples/seatunnel/clone/copy-schema.py" + spec = importlib.util.spec_from_file_location("copy_schema", path) + module = importlib.util.module_from_spec(spec) + spec.loader.exec_module(module) + for label_type in ("PARENT", "SUB", "UNKNOWN"): + with self.subTest(label_type=label_type): + source = mock.Mock(return_value={"edgelabels": [ + {"name": "relation", "edgelabel_type": label_type}]}) + target = mock.Mock() + with mock.patch.object(module, "client", side_effect=[source, target]) as factory, \ + mock.patch("sys.argv", [str(path), "--source", "http://source/graph", + "--target", "http://target/graph"]): + with self.assertRaisesRegex(SystemExit, "only supports NORMAL edge labels"): + module.main() + source.assert_called_once_with("GET", "/schema") + factory.assert_called_once_with("http://source/graph", "SOURCE") + target.assert_not_called() + + def test_legacy_release_without_seatunnel_values_still_renders(self): + with tempfile.TemporaryDirectory(prefix="seatunnel-legacy-") as directory: + chart = Path(directory) / "chart" + shutil.copytree(CHART, chart) + (chart / "values.yaml").write_bytes((CHART / "testdata/values-pre-hardening.yaml").read_bytes()) + for arguments in ([], ["--is-upgrade"], ["--set", "global.imageRegistry=registry.example.com"]): + with self.subTest(arguments=arguments): + result = self.run_helm(["template", "legacy", str(chart), *arguments], + HELM_TEMPLATE_TIMEOUT_SECONDS) + self.assertEqual(0, result.returncode, result.stderr) + documents = [doc for doc in yaml.safe_load_all(result.stdout) if doc] + self.assertEqual([], engine_resources(documents)) + # An earlier enabled release also predates the entry/auth fields. + previous = yaml.safe_load((CHART / "values.yaml").read_text()) + previous["seatunnel"]["enabled"] = True + for field in ("service", "auth"): + previous["seatunnel"].pop(field) + (chart / "values.yaml").write_text(yaml.safe_dump(previous)) + documents = self.documents(chart=chart) + http = yaml.safe_load(engine_config(documents)["seatunnel.yaml"])["seatunnel"]["engine"]["http"] + self.assertTrue(http["enable-basic-auth"]) + self.assertEqual("admin", http["basic-auth-username"]) + + def test_modes_share_one_native_hybrid_workload_and_services(self): + modes = [self.documents({}), self.documents({"mode": "cluster", "replicas": 3})] + workloads = [] + for documents, replicas, backups in zip(modes, (1, 3), (0, 1)): + workload, = engine_resources(documents, "StatefulSet") + workloads.append(workload) + self.assertEqual([], engine_resources(documents, "Deployment")) + self.assertEqual(replicas, workload["spec"]["replicas"]) + container, = pod(workload)["spec"]["containers"] + launch = " ".join(container.get("command", []) + container.get("args", [])) + self.assertIn("/opt/seatunnel/bin/seatunnel-cluster.sh", launch) + self.assertNotIn(" -r ", launch) + data = engine_config(documents) + for option in ("-Xms512m", "-Xmx1g", "-XX:+UseG1GC", "-XX:MaxMetaspaceSize=256m", + "-XX:+HeapDumpOnOutOfMemoryError", "-XX:HeapDumpPath=/tmp/seatunnel.hprof"): + self.assertIn(option, data["jvm_options"].splitlines()) + engine = yaml.safe_load(data["seatunnel.yaml"])["seatunnel"]["engine"] + self.assertEqual(backups, engine["backup-count"]) + self.assertTrue(engine["http"]["enable-http"]) + self.assertEqual(8080, engine["http"]["port"]) + self.assertEqual(workloads[0]["metadata"]["name"], workloads[1]["metadata"]["name"]) + self.assertEqual(workloads[0]["spec"]["selector"], workloads[1]["spec"]["selector"]) + services = lambda docs: {doc["metadata"]["name"]: doc["spec"]["selector"] + for doc in engine_resources(docs, "Service")} + self.assertEqual(services(modes[0]), services(modes[1])) + + def test_dns_discovery_and_services_are_scoped_to_engine_and_release(self): + documents = self.documents({"mode": "cluster"}, namespace="graphs", fullnameOverride="transfer") + workload, = engine_resources(documents, "StatefulSet") + labels = pod(workload)["metadata"]["labels"] + services = engine_resources(documents, "Service") + headless, = [service for service in services if service["spec"].get("clusterIP") == "None"] + api, = [service for service in services if service is not headless] + self.assertEqual("transfer-seatunnel", workload["metadata"]["name"]) + self.assertEqual(headless["metadata"]["name"], workload["spec"]["serviceName"]) + self.assertTrue(headless["spec"]["publishNotReadyAddresses"]) + self.assertIn(8080, [port["port"] for port in api["spec"]["ports"]]) + for service in services: + selector = service["spec"]["selector"] + self.assertTrue(matches(selector, labels)) + self.assertFalse(matches(selector, {**labels, "app.kubernetes.io/instance": "other"})) + for other in documents: + if other["kind"] in ("Deployment", "StatefulSet") and other is not workload: + self.assertFalse(matches(selector, pod(other)["metadata"]["labels"])) + data = engine_config(documents) + discovery = yaml.safe_load(data["hazelcast.yaml"])["hazelcast"]["network"]["join"]["kubernetes"] + self.assertTrue(discovery["enabled"]) + self.assertTrue(discovery["service-dns"].startswith(headless["metadata"]["name"] + ".graphs.svc")) + self.assertNotIn("service-name", discovery) + # DNS discovery otherwise dials Hazelcast's default5701 and creates + # isolated one-member clusters even while both Pods report Ready. + member_port = next(port["containerPort"] for port in + pod(workload)["spec"]["containers"][0]["ports"] + if port["name"] == "hazelcast") + self.assertEqual(member_port, discovery["service-port"]) + client = yaml.safe_load(data["hazelcast-client.yaml"])["hazelcast-client"] + self.assertIn(discovery["service-dns"] + ":5801", client["network"]["cluster-members"]) + self.assertFalse(any(doc["kind"] in ("Role", "RoleBinding") for doc in engine_resources(documents))) + + def test_http_entry_inherits_mode_and_preserves_explicit_service_settings(self): + for mode, service_type in (("easy", "NodePort"), ("strict", "ClusterIP")): + with self.subTest(mode=mode): + documents = self.documents({}, mode=mode) + api, = [doc for doc in engine_resources(documents, "Service") + if doc["spec"].get("clusterIP") != "None"] + self.assertEqual(service_type, api["spec"]["type"]) + self.assertNotIn("nodePort", api["spec"]["ports"][0]) + for mode, service in (("easy", {"type": "ClusterIP"}), + ("strict", {"type": "NodePort", "nodePort": 31081})): + with self.subTest(mode=mode, service=service): + documents = self.documents({"service": service}, mode=mode) + headless, = [doc for doc in engine_resources(documents, "Service") + if doc["spec"].get("clusterIP") == "None"] + api, = [doc for doc in engine_resources(documents, "Service") if doc is not headless] + self.assertEqual(service["type"], api["spec"]["type"]) + if "nodePort" in service: + self.assertEqual(service["nodePort"], api["spec"]["ports"][0]["nodePort"]) + self.assertNotIn("nodePort", headless["spec"]["ports"][0]) + result = self.render({"service": {"type": "ClusterIP", "nodePort": 31081}}) + self.assertNotEqual(0, result.returncode) + + def test_local_graph_defaults_follow_rendered_server_and_admin_secret(self): + documents = self.documents({}, namespace="graphs", release="fixture-" + "r" * 40, + fullnameOverride="graph-copy-" + "n" * 55, + server={"port": 8182, "graphSpace": "tenant_a"}) + server, = [doc for doc in documents if doc["kind"] == "Service" + and doc["metadata"]["labels"].get("app.kubernetes.io/component") == "server" + and doc["spec"].get("clusterIP") != "None"] + server_workload, = [doc for doc in documents if doc["kind"] == "Deployment" + and doc["metadata"]["labels"].get("app.kubernetes.io/component") == "server"] + server_container = next(entry for entry in pod(server_workload)["spec"]["containers"] + if entry["name"] == "server") + server_password = next(entry["valueFrom"] for entry in server_container["env"] + if entry["name"] == "PASSWORD") + workload, = engine_resources(documents, "StatefulSet") + container, = pod(workload)["spec"]["containers"] + env = {entry["name"]: entry for entry in container["env"]} + self.assertEqual(server["metadata"]["name"] + ".graphs.svc", env["HG_HOST"]["value"]) + self.assertEqual(str(server["spec"]["ports"][0]["port"]), env["HG_PORT"]["value"]) + self.assertEqual("http", env["HG_PROTOCOL"]["value"]) + self.assertEqual("tenant_a", env["HG_GRAPH_SPACE"]["value"]) + self.assertEqual("graph_a", env["HG_GRAPH_NAME"]["value"]) + self.assertEqual("admin", env["HG_USERNAME"]["value"]) + self.assertEqual(server_password, env["HG_PASSWORD"]["valueFrom"]) + + def test_existing_admin_secret_and_disabled_auth_reach_local_graph_environment(self): + cases = (({"enabled": True, "admin": {"existingSecret": "graph-admin", "key": "admin-pass"}}, + {"name": "HG_PASSWORD", "valueFrom": { + "secretKeyRef": {"name": "graph-admin", "key": "admin-pass"}}}), + ({"enabled": False}, {"name": "HG_PASSWORD", "value": ""})) + for auth, password in cases: + with self.subTest(auth=auth): + documents = self.documents({}, server={"auth": auth}, hubble={"enabled": False}) + workload, = engine_resources(documents, "StatefulSet") + container, = pod(workload)["spec"]["containers"] + self.assertIn(password, container["env"]) + + def test_explicit_environment_overrides_defaults_without_duplicate_names(self): + overrides = [{"name": "HG_HOST", "value": "external-graph.example.com"}, + {"name": "HG_PORT", "value": "8443"}, + {"name": "HG_PROTOCOL", "value": "https"}, + {"name": "HG_GRAPH_SPACE", "value": "other"}, + {"name": "HG_GRAPH_NAME", "value": "existing_graph"}, + {"name": "HG_USERNAME", "value": "writer"}, + {"name": "HG_PASSWORD", "valueFrom": { + "secretKeyRef": {"name": "writer-auth", "key": "password"}}}, + {"name": "TZ", "value": "UTC"}] + documents = self.documents({"env": overrides}) + workload, = engine_resources(documents, "StatefulSet") + container, = pod(workload)["spec"]["containers"] + env = {entry["name"]: entry for entry in container["env"]} + self.assertEqual(len(env), len(container["env"])) + for expected in overrides: + self.assertEqual(expected, env[expected["name"]]) + + def test_demo_jobs_are_mounted_and_keep_credentials_as_placeholders(self): + password = "render-test-password" + documents = self.documents({}, server={"auth": {"admin": {"password": password}}}) + configmaps = engine_resources(documents, "ConfigMap") + jobs = {doc["metadata"]["name"]: doc["data"] for doc in configmaps + if any(filename.endswith(".conf") for filename in doc["data"])} + self.assertEqual(2, sum(filename.endswith(".conf") for data in jobs.values() for filename in data)) + workload, = engine_resources(documents, "StatefulSet") + spec = pod(workload)["spec"] + container, = spec["containers"] + for configmap_name, data in jobs.items(): + volume_names = {volume["name"] for volume in spec["volumes"] + if volume.get("configMap", {}).get("name") == configmap_name} + self.assertTrue(volume_names) + mounts = [mount for mount in container["volumeMounts"] if mount["name"] in volume_names] + self.assertTrue(mounts) + for filename, content in data.items(): + if filename.endswith(".conf"): + self.assertTrue(any(mount["mountPath"] == "/opt/seatunnel/config/" + filename + for mount in mounts)) + for name in ("HG_HOST", "HG_PORT", "HG_PROTOCOL", "HG_GRAPH_SPACE", "HG_GRAPH_NAME", + "HG_USERNAME", "HG_PASSWORD"): + self.assertIn("${" + name + "}", content) + for data in (doc["data"] for doc in configmaps): + self.assertFalse(any(password in content for content in data.values())) + + def test_seatunnel_admin_is_generated_separately_and_kept_out_of_public_configuration(self): + graph_password = "graph-only-admin-password" + documents = self.documents({"auth": {"password": ""}}, + server={"auth": {"admin": {"password": graph_password}}}) + secret = engine_auth_secret(documents) + username = base64.b64decode(secret["data"]["username"]).decode() + password = base64.b64decode(secret["data"]["password"]).decode() + self.assertEqual("admin", username) + self.assertGreaterEqual(len(password), 32) + self.assertNotEqual(graph_password, password) + http = yaml.safe_load(engine_config(documents)["seatunnel.yaml"])["seatunnel"]["engine"]["http"] + self.assertTrue(http["enable-basic-auth"]) + self.assertEqual(username, http["basic-auth-username"]) + self.assertEqual(password, http["basic-auth-password"]) + for configmap in engine_resources(documents, "ConfigMap"): + self.assertNotIn("seatunnel.yaml", configmap["data"]) + self.assertNotIn(password, yaml.safe_dump(configmap["data"])) + workload, = engine_resources(documents, "StatefulSet") + self.assertNotIn(password, yaml.safe_dump(pod(workload))) + volumes = {volume["name"] for volume in pod(workload)["spec"]["volumes"] + if volume.get("secret", {}).get("secretName") == secret["metadata"]["name"]} + container, = pod(workload)["spec"]["containers"] + self.assertTrue(any(mount["name"] in volumes + and mount["mountPath"] == "/opt/seatunnel/config/seatunnel.yaml" + for mount in container["volumeMounts"])) + + def test_native_basic_auth_preserves_special_credentials_and_cannot_be_disabled_by_config(self): + username = "ops'\"\\$" + password = "quote'\" backslash\\ dollar$ colon:" + native = {"seatunnel": {"engine": {"backup-count": 0, "http": { + "enable-http": True, "port": 8080, "enable-basic-auth": False, + "basic-auth-username": "ignored-user", "basic-auth-password": "ignored-password"}}}} + documents = self.documents({"auth": {"username": username, "password": password}, + "config": {"seatunnel.yaml": yaml.safe_dump(native)}}) + secret = engine_auth_secret(documents) + self.assertEqual(username, base64.b64decode(secret["data"]["username"]).decode()) + self.assertEqual(password, base64.b64decode(secret["data"]["password"]).decode()) + http = yaml.safe_load(engine_config(documents)["seatunnel.yaml"])["seatunnel"]["engine"]["http"] + self.assertTrue(http["enable-basic-auth"]) + self.assertEqual(username, http["basic-auth-username"]) + self.assertEqual(password, http["basic-auth-password"]) + + def test_image_secret_environment_and_native_file_overrides_reach_engine(self): + custom = {"jvm_options": "-Xms128m\n-Xmx384m\n", "log4j2.properties": "status=warn\n"} + secret_env = {"name": "HG_PASSWORD", "valueFrom": {"secretKeyRef": {"name": "graph-auth", "key": "password"}}} + documents = self.documents({"image": {"repository": "team/seatunnel", "tag": "tested", "pullPolicy": "Never"}, + "env": [secret_env], "config": custom}, + **{"global": {"imageRegistry": "registry.example.com"}, + "imagePullSecrets": [{"name": "registry-auth"}]}) + workload, = engine_resources(documents, "StatefulSet") + spec = pod(workload)["spec"] + container, = spec["containers"] + self.assertEqual("registry.example.com/team/seatunnel:tested", container["image"]) + self.assertEqual("Never", container["imagePullPolicy"]) + self.assertEqual([{"name": "registry-auth"}], spec["imagePullSecrets"]) + self.assertIn(secret_env, container["env"]) + data = engine_config(documents) + mounts = {entry["subPath"] for entry in container["volumeMounts"]} + for filename, content in custom.items(): + self.assertEqual(content, data[filename]) + self.assertIn(filename, mounts) + previous, = engine_resources(self.documents({}), "StatefulSet") + self.assertNotEqual(pod(previous)["metadata"]["annotations"]["checksum/config"], + pod(workload)["metadata"]["annotations"]["checksum/config"]) + + def test_invalid_enabled_mode_and_cluster_size_fail_before_install(self): + for values in ({"mode": "invalid"}, {"mode": "cluster", "replicas": 0}, + {"mode": "cluster", "replicas": 1}, + {"auth": {"username": "${user.name}"}}, + {"auth": {"password": "${user.name}"}}, + {"auth": {"username": " admin"}}, + {"auth": {"password": "password "}}, + {"auth": {"username": "admin:operator"}}): + with self.subTest(values=values): + self.assertNotEqual(0, self.render(values).returncode) + + def test_strict_policies_allow_engine_server_dns_and_native_extensions(self): + extra_in = {"from": [{"ipBlock": {"cidr": "10.10.0.0/16"}}], "ports": [{"port": 8080}]} + extra_out = {"to": [{"ipBlock": {"cidr": "10.20.0.0/16"}}], "ports": [{"port": 9099}]} + documents = self.documents({"mode": "cluster"}, mode="strict", server={"port": 8182}, + networkPolicy={"seatunnel": {"extraIngress": [extra_in], "extraEgress": [extra_out]}}) + policies = [doc for doc in documents if doc["kind"] == "NetworkPolicy"] + engine = next(doc["spec"] for doc in policies if doc["metadata"]["name"].endswith("-seatunnel")) + server = next(doc["spec"] for doc in policies if doc["metadata"]["name"].endswith("-server")) + workload, = engine_resources(documents, "StatefulSet") + self.assertTrue(matches(engine["podSelector"]["matchLabels"], pod(workload)["metadata"]["labels"])) + self.assertIn(extra_in, engine["ingress"]) + self.assertIn(extra_out, engine["egress"]) + + def allows(rules, direction, selector, port): + return any(peer.get("podSelector", {}).get("matchLabels") == selector + and any(entry["port"] == port for entry in rule.get("ports", [])) + for rule in rules for peer in rule.get(direction, [])) + + selector = engine["podSelector"]["matchLabels"] + self.assertTrue(allows(server["ingress"], "from", selector, 8182)) + self.assertTrue(allows(engine["egress"], "to", server["podSelector"]["matchLabels"], 8182)) + for direction, peer_key in (("ingress", "from"), ("egress", "to")): + for port in (5801, 8080): + self.assertTrue(allows(engine[direction], peer_key, selector, port)) + dns = [port for rule in engine["egress"] if not rule.get("to") for port in rule["ports"]] + self.assertTrue({("UDP", 53), ("TCP", 53)} <= {(p["protocol"], p["port"]) for p in dns}) + + def test_member_ingress_is_isolated_when_full_network_policy_is_disabled(self): + for mode in ("easy", "strict"): + with self.subTest(mode=mode): + documents = self.documents({}, mode=mode, networkPolicy={"enabled": False}) + policy, = engine_resources(documents, "NetworkPolicy") + spec = policy["spec"] + self.assertEqual(["Ingress"], spec["policyTypes"]) + self.assertNotIn("egress", spec) + workload, = engine_resources(documents, "StatefulSet") + labels = pod(workload)["metadata"]["labels"] + self.assertTrue(matches(spec["podSelector"]["matchLabels"], labels)) + member_rules = [rule for rule in spec["ingress"] + if any(port["port"] == 5801 for port in rule["ports"])] + self.assertEqual(1, len(member_rules)) + self.assertEqual([{"podSelector": {"matchLabels": spec["podSelector"]["matchLabels"]}}], + member_rules[0]["from"]) + selector = member_rules[0]["from"][0]["podSelector"]["matchLabels"] + self.assertFalse(matches(selector, {**labels, "app.kubernetes.io/component": "server"})) + self.assertFalse(matches(selector, {**labels, "app.kubernetes.io/instance": "other"})) + open_rules = [rule for rule in spec["ingress"] if not rule.get("from")] + self.assertEqual([{"ports": [{"protocol": "TCP", "port": 8080}]}], open_rules) + + def test_engine_security_defaults_and_native_scheduling_overrides(self): + workload, = engine_resources(self.documents({}), "StatefulSet") + container, = pod(workload)["spec"]["containers"] + self.assertEqual({"allowPrivilegeEscalation": False, + "capabilities": {"drop": ["ALL"]}, + "seccompProfile": {"type": "RuntimeDefault"}}, container["securityContext"]) + overrides = { + "podSecurityContext": {"runAsUser": 1000, "fsGroup": 1000}, + "securityContext": {"runAsNonRoot": True}, + "nodeSelector": {"workload": "ingestion"}, + "tolerations": [{"key": "ingestion", "operator": "Exists", "effect": "NoSchedule"}], + "affinity": {"podAntiAffinity": {"preferredDuringSchedulingIgnoredDuringExecution": []}}, + "priorityClassName": "ingestion", + } + workload, = engine_resources(self.documents(overrides), "StatefulSet") + spec = pod(workload)["spec"] + self.assertEqual(overrides["podSecurityContext"], spec["securityContext"]) + for field in ("nodeSelector", "tolerations", "affinity", "priorityClassName"): + self.assertEqual(overrides[field], spec[field]) + container, = spec["containers"] + self.assertTrue(container["securityContext"]["runAsNonRoot"]) + self.assertFalse(container["securityContext"]["allowPrivilegeEscalation"]) + for invalid in ({"securityContext": []}, {"nodeSelector": []}, {"tolerations": {}}, + {"affinity": []}, {"priorityClassName": 1}): + with self.subTest(invalid=invalid): + self.assertNotEqual(0, self.render(invalid).returncode) + + def test_packaged_chart_has_both_modes_without_seatunnel_dependency(self): + with tempfile.TemporaryDirectory(prefix="seatunnel-package-") as directory: + self.run_helm(["package", str(CHART), "--destination", directory], + HELM_PACKAGE_TIMEOUT_SECONDS, check=True) + package = next(Path(directory).glob("*.tgz")) + with tarfile.open(package, "r:gz") as archive: + metadata = yaml.safe_load(archive.extractfile("hugegraph-helm/Chart.yaml")) + self.assertFalse(any("seatunnel" in dependency["name"] + for dependency in metadata.get("dependencies", []))) + self.assertTrue(any(name.endswith("/templates/seatunnel.yaml") for name in archive.getnames())) + self.assertEqual([], engine_resources(self.documents(chart=package))) + for mode, replicas in (("standalone", 1), ("cluster", 2)): + documents = self.documents({"mode": mode}, chart=package) + workload, = engine_resources(documents, "StatefulSet") + self.assertEqual(replicas, workload["spec"]["replicas"]) + source, = engine_resources(self.documents({"mode": mode}), "StatefulSet") + self.assertEqual(source["spec"], workload["spec"]) + + +if __name__ == "__main__": + unittest.main()