This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/dev/pr-12885-619cdabf2b4f71169a2b6c6b2f7437075c7c7f3e in repository https://gitbox.apache.org/repos/asf/seatunnel.git
commit 7604a83ec802a0b148e98e557fb9184c48127988 Author: zhiweiniu <[email protected]> AuthorDate: Sun Oct 25 05:34:13 2026 +0000 [Feature][Zeta] Support environment-variable object storage credential (#12885) --- docs/en/engines/zeta/checkpoint-storage.md | 63 ++++++ docs/en/engines/zeta/hybrid-cluster-deployment.md | 21 +- .../engines/zeta/separated-cluster-deployment.md | 21 +- docs/en/engines/zeta/state-storage-and-recovery.md | 2 + .../en/getting-started/kubernetes/configuration.md | 66 ++++++- docs/zh/engines/zeta/checkpoint-storage.md | 57 ++++++ docs/zh/engines/zeta/hybrid-cluster-deployment.md | 16 +- .../engines/zeta/separated-cluster-deployment.md | 16 +- docs/zh/engines/zeta/state-storage-and-recovery.md | 2 + .../zh/getting-started/kubernetes/configuration.md | 66 ++++++- .../checkpoint-storage-hdfs/pom.xml | 17 +- .../EnvironmentCredentialConfigurationTest.java | 220 +++++++++++++++++++++ .../imap-storage-plugins/imap-storage-file/pom.xml | 13 +- .../EnvironmentCredentialConfigurationTest.java | 130 ++++++++++++ 14 files changed, 683 insertions(+), 27 deletions(-) diff --git a/docs/en/engines/zeta/checkpoint-storage.md b/docs/en/engines/zeta/checkpoint-storage.md index b3db813485..d366382f62 100644 --- a/docs/en/engines/zeta/checkpoint-storage.md +++ b/docs/en/engines/zeta/checkpoint-storage.md @@ -43,6 +43,27 @@ seatunnel: Notice: namespace must end with "/". +#### Credentials from environment variables + +Object storage credentials can be supplied by the cloud SDK from environment variables. Set +the corresponding credential provider key and do not put static access keys or secret keys in +`seatunnel.yaml`: + +| `storage.type` | Credential provider key | Provider class | Environment variables | +| --- | --- | --- | --- | +| `s3` | `fs.s3a.aws.credentials.provider` | `com.amazonaws.auth.EnvironmentVariableCredentialsProvider` | `AWS_ACCESS_KEY_ID`, `AWS_SECRET_ACCESS_KEY`, optional `AWS_SESSION_TOKEN` | +| `oss` | `fs.oss.credentials.provider` | `com.aliyun.oss.common.auth.EnvironmentVariableCredentialsProvider` | `OSS_ACCESS_KEY_ID`, `OSS_ACCESS_KEY_SECRET`, optional `OSS_SESSION_TOKEN` | +| `cos` | `fs.cosn.credentials.provider` | `org.apache.hadoop.fs.cosn.auth.EnvironmentVariableCredentialsProvider` | `COSN_SECRET_ID`, `COSN_SECRET_KEY` | + +The environment variables and provider implementations are fixed by the underlying Hadoop +filesystem and cloud SDK. Inject the variables into every SeaTunnel process that accesses +checkpoint storage. Existing static credential and cloud-native identity configurations remain +supported. + +GCS uses Application Default Credentials instead of a credential provider key. Configure +`GOOGLE_APPLICATION_CREDENTIALS` with a mounted key file when needed, or rely on Workload Identity +and other ADC mechanisms. + #### OSS Aliyun OSS based hdfs-file you can refer [Hadoop OSS Docs](https://hadoop.apache.org/docs/stable/hadoop-aliyun/tools/hadoop-aliyun/index.html) to config oss. @@ -70,6 +91,24 @@ seatunnel: fs.oss.endpoint: endpoint address ``` +To use credentials from environment variables, remove `fs.oss.accessKeyId` and +`fs.oss.accessKeySecret` and select the environment variable provider. `OSS_ACCESS_KEY_ID` and +`OSS_ACCESS_KEY_SECRET` must be available to every master and worker: + +```yaml +seatunnel: + engine: + checkpoint: + storage: + type: hdfs + plugin-config: + namespace: /seatunnel/checkpoint/ + storage.type: oss + oss.bucket: oss://your-bucket + fs.oss.endpoint: oss-cn-hangzhou.aliyuncs.com + fs.oss.credentials.provider: com.aliyun.oss.common.auth.EnvironmentVariableCredentialsProvider +``` + For additional reading on the Hadoop Credential Provider API, you can see: [Credential Provider API](https://hadoop.apache.org/docs/stable/hadoop-project-dist/hadoop-common/CredentialProviderAPI.html). For Aliyun OSS Credential Provider implements, you can see: [Auth Credential Providers](https://github.com/aliyun/aliyun-oss-java-sdk/tree/master/src/main/java/com/aliyun/oss/common/auth) @@ -102,6 +141,18 @@ seatunnel: fs.cosn.bucket.region: your-region ``` +To use credentials from environment variables, remove the static `secretId` and `secretKey` +entries and configure: + +```yaml + storage.type: cos + cos.bucket: cosn://your-bucket + fs.cosn.bucket.region: your-region + fs.cosn.credentials.provider: org.apache.hadoop.fs.cosn.auth.EnvironmentVariableCredentialsProvider +``` + +`COSN_SECRET_ID` and `COSN_SECRET_KEY` must be available to every master and worker. + For additional reading on the Hadoop Credential Provider API, you can see: [Credential Provider API](https://hadoop.apache.org/docs/stable/hadoop-project-dist/hadoop-common/CredentialProviderAPI.html). For additional COS configuration, you can see: [Tencent Hadoop-COS Docs](https://doc.fincloud.tencent.cn/tcloud/Storage/COS/846365/hadoop) @@ -188,6 +239,18 @@ seatunnel: ``` +To read credentials from environment variables instead, remove `fs.s3a.access.key` and +`fs.s3a.secret.key` and configure: + +```yaml + storage.type: s3 + s3.bucket: s3a://your-bucket + fs.s3a.aws.credentials.provider: com.amazonaws.auth.EnvironmentVariableCredentialsProvider +``` + +`AWS_ACCESS_KEY_ID` and `AWS_SECRET_ACCESS_KEY` must be available to every master and worker. +Temporary credentials additionally use `AWS_SESSION_TOKEN`. + If you used `InstanceProfileCredentialsProvider`, which supports use of instance profile credentials if running in an EC2 VM, you can check [iam-roles-for-amazon-ec2](https://docs.aws.amazon.com/zh_cn/AWSEC2/latest/UserGuide/iam-roles-for-amazon-ec2.html). You can config like this: diff --git a/docs/en/engines/zeta/hybrid-cluster-deployment.md b/docs/en/engines/zeta/hybrid-cluster-deployment.md index e6cb31935d..abf1489d11 100644 --- a/docs/en/engines/zeta/hybrid-cluster-deployment.md +++ b/docs/en/engines/zeta/hybrid-cluster-deployment.md @@ -364,14 +364,24 @@ map: fs.oss.endpoint: OSS endpoint ``` +To use OSS credentials from environment variables, remove `fs.oss.accessKeyId` and +`fs.oss.accessKeySecret` and add: + +```yaml + fs.oss.credentials.provider: com.aliyun.oss.common.auth.EnvironmentVariableCredentialsProvider +``` + +The provider reads `OSS_ACCESS_KEY_ID` and `OSS_ACCESS_KEY_SECRET`, plus optional +`OSS_SESSION_TOKEN`. + Notice: When using OSS, make sure that the following jars are in the lib directory. The `seatunnel-shade-hadoop3-uber` JAR comes from the [Apache SeaTunnel Shade](https://github.com/apache/seatunnel-shade) project, which provides a shaded (package-relocated) version of the Hadoop client. All third-party classes are relocated under `org.apache.seatunnel.shade.*` to avoid classpath conflicts with SeaTunnel's own dependencies. The version follows the `${library.version}-${seatunnel.shade.version}` format (e.g., `3.1.4-3.0.0`). Refer to the actual JAR file name in your SeaT [...] ``` -aliyun-sdk-oss-3.13.2.jar -hadoop-aliyun-3.3.6.jar -jdom2-2.0.6.jar +aliyun-sdk-oss-3.4.1.jar +hadoop-aliyun-3.1.4.jar +jdom-1.1.jar netty-buffer-4.1.89.Final.jar netty-common-4.1.89.Final.jar seatunnel-shade-hadoop3-uber-${seatunnel.shade.hadoop.version}-${seatunnel.shade.version}.jar @@ -405,6 +415,11 @@ map: fs.s3a.aws.credentials.provider: org.apache.hadoop.fs.s3a.SimpleAWSCredentialsProvider ``` +For environment-variable credentials, remove `fs.s3a.access.key` and `fs.s3a.secret.key` and use +`fs.s3a.aws.credentials.provider: com.amazonaws.auth.EnvironmentVariableCredentialsProvider`. +The provider reads `AWS_ACCESS_KEY_ID` and `AWS_SECRET_ACCESS_KEY`, plus optional +`AWS_SESSION_TOKEN`. + Notice: When using S3, make sure that the following jars are in the lib directory. Both JARs come from the [Apache SeaTunnel Shade](https://github.com/apache/seatunnel-shade) project: diff --git a/docs/en/engines/zeta/separated-cluster-deployment.md b/docs/en/engines/zeta/separated-cluster-deployment.md index a04d6f8074..6eee00d7c7 100644 --- a/docs/en/engines/zeta/separated-cluster-deployment.md +++ b/docs/en/engines/zeta/separated-cluster-deployment.md @@ -316,14 +316,24 @@ map: fs.oss.endpoint: OSS endpoint ``` +To use OSS credentials from environment variables, remove `fs.oss.accessKeyId` and +`fs.oss.accessKeySecret` and add: + +```yaml + fs.oss.credentials.provider: com.aliyun.oss.common.auth.EnvironmentVariableCredentialsProvider +``` + +The provider reads `OSS_ACCESS_KEY_ID` and `OSS_ACCESS_KEY_SECRET`, plus optional +`OSS_SESSION_TOKEN`. + Notice: When using OSS, make sure that the following jars are in the lib directory. The `seatunnel-shade-hadoop3-uber` JAR comes from the [Apache SeaTunnel Shade](https://github.com/apache/seatunnel-shade) project, which provides a shaded (package-relocated) version of the Hadoop client. All third-party classes are relocated under `org.apache.seatunnel.shade.*` to avoid classpath conflicts with SeaTunnel's own dependencies. The version follows the `${library.version}-${seatunnel.shade.version}` format (e.g., `3.1.4-3.0.0`). Refer to the actual JAR file name in your SeaT [...] ``` -aliyun-sdk-oss-3.13.2.jar -hadoop-aliyun-3.3.6.jar -jdom2-2.0.6.jar +aliyun-sdk-oss-3.4.1.jar +hadoop-aliyun-3.1.4.jar +jdom-1.1.jar netty-buffer-4.1.89.Final.jar netty-common-4.1.89.Final.jar seatunnel-shade-hadoop3-uber-${seatunnel.shade.hadoop.version}-${seatunnel.shade.version}.jar @@ -356,6 +366,11 @@ map: fs.s3a.aws.credentials.provider: org.apache.hadoop.fs.s3a.SimpleAWSCredentialsProvider ``` +For environment-variable credentials, remove `fs.s3a.access.key` and `fs.s3a.secret.key` and use +`fs.s3a.aws.credentials.provider: com.amazonaws.auth.EnvironmentVariableCredentialsProvider`. +The provider reads `AWS_ACCESS_KEY_ID` and `AWS_SECRET_ACCESS_KEY`, plus optional +`AWS_SESSION_TOKEN`. + Notice: When using S3, make sure that the following jars are in the lib directory. Both JARs come from the [Apache SeaTunnel Shade](https://github.com/apache/seatunnel-shade) project: diff --git a/docs/en/engines/zeta/state-storage-and-recovery.md b/docs/en/engines/zeta/state-storage-and-recovery.md index 526fd50b0b..b0a7460cb1 100644 --- a/docs/en/engines/zeta/state-storage-and-recovery.md +++ b/docs/en/engines/zeta/state-storage-and-recovery.md @@ -60,6 +60,8 @@ seatunnel: # fs.s3a.endpoint: https://s3.amazonaws.com # fs.s3a.access.key: <your-access-key> # fs.s3a.secret.key: <your-secret-key> + # Environment credentials: + # fs.s3a.aws.credentials.provider: com.amazonaws.auth.EnvironmentVariableCredentialsProvider ``` ### Checkpoint retention and cleanup diff --git a/docs/en/getting-started/kubernetes/configuration.md b/docs/en/getting-started/kubernetes/configuration.md index f3faa557d5..94a9888866 100644 --- a/docs/en/getting-started/kubernetes/configuration.md +++ b/docs/en/getting-started/kubernetes/configuration.md @@ -283,7 +283,20 @@ properties: Alibaba Cloud OSS uses Hadoop OSS configuration and selects the target bucket with `oss.bucket`. Set `fs.oss.endpoint` to the actual endpoint of the bucket region, private cloud, or OSS-compatible service. -The default Hadoop Aliyun OSS credential provider reads `fs.oss.accessKeyId` and `fs.oss.accessKeySecret`. Temporary credentials can additionally set `fs.oss.securityToken`. It does not provide a default environment-variable credential chain like Hadoop S3A, so credentials injected from Kubernetes Secret as environment variables are not used automatically. +The default Hadoop Aliyun OSS credential provider reads `fs.oss.accessKeyId` and `fs.oss.accessKeySecret`. Temporary credentials can additionally set `fs.oss.securityToken`. To read credentials from environment variables and avoid rendering them into the ConfigMap, select the OSS SDK environment variable provider: + +```yaml +properties: + type: hdfs + namespace: /seatunnel/imap + clusterName: seatunnel-cluster + storage.type: oss + oss.bucket: oss://seatunnel-bucket + fs.oss.endpoint: https://oss-<region>.aliyuncs.com + fs.oss.credentials.provider: com.aliyun.oss.common.auth.EnvironmentVariableCredentialsProvider +``` + +The provider reads the fixed environment variables `OSS_ACCESS_KEY_ID` and `OSS_ACCESS_KEY_SECRET`, plus `OSS_SESSION_TOKEN` for temporary credentials. Inject them into every Master and Worker Pod. ```yaml hazelcast: @@ -304,12 +317,59 @@ hazelcast: fs.oss.accessKeySecret: YOUR_ACCESS_KEY_SECRET ``` -If you do not want to render OSS credentials into the final ConfigMap, use Hadoop credential provider for `fs.oss.accessKeyId` and `fs.oss.accessKeySecret`, or set `fs.oss.credentials.provider` to a custom provider. The custom provider must implement the Aliyun OSS SDK `com.aliyun.oss.common.auth.CredentialsProvider` interface and must be available in the SeaTunnel image. +If you do not want to render OSS credentials into the final ConfigMap, use the environment variable provider shown above, a Hadoop credential provider, or a custom `fs.oss.credentials.provider` implementation. A custom provider must implement the Aliyun OSS SDK `com.aliyun.oss.common.auth.CredentialsProvider` interface and must be available in the SeaTunnel image. + +#### Inject credentials from Kubernetes Secrets + +Kubernetes injects Secret values into the container process environment when the Pod starts. The values are not written to the Docker image, and SeaTunnel does not need a Dockerfile `ENV` entry. Inject the credentials into every Master and Worker container that can access checkpoint or IMap storage. + +If the Secret keys already use the fixed provider variable names, use `envFrom`: + +```yaml +apiVersion: v1 +kind: Secret +metadata: + name: seatunnel-oss-credentials +type: Opaque +stringData: + OSS_ACCESS_KEY_ID: your-access-key + OSS_ACCESS_KEY_SECRET: your-secret-key + OSS_SESSION_TOKEN: your-session-token # omit for permanent credentials +--- +spec: + template: + spec: + containers: + - name: seatunnel + envFrom: + - secretRef: + name: seatunnel-oss-credentials +``` + +If the Secret uses different key names, map them explicitly so the provider receives the fixed names: + +```yaml +containers: + - name: seatunnel + env: + - name: OSS_ACCESS_KEY_ID + valueFrom: + secretKeyRef: + name: seatunnel-oss-credentials + key: access-key + - name: OSS_ACCESS_KEY_SECRET + valueFrom: + secretKeyRef: + name: seatunnel-oss-credentials + key: secret-key +``` + +Do not set a prefix on `envFrom`, because the provider expects the exact variable names shown above. Environment variables in a running container do not update when the Secret changes; restart or roll out the StatefulSet after rotating credentials. When `backup-count: 1` is used, at least 2 Master replicas are required to store IMap backup replicas. :::caution Warning -Do not commit real access keys or secret keys to ConfigMaps or Git repositories. Kubernetes does not resolve `secretKeyRef` inside ConfigMap file content. SeaTunnel and Hazelcast must read a final, usable YAML configuration. In production, generate the final ConfigMap through Helm, Kustomize, External Secrets, CI/CD, or the image build process, or use credential mechanisms actually supported by the underlying filesystem, such as Hadoop credential provider, AWS SDK environment variable pr [...] +Do not commit real access keys or secret keys to ConfigMaps or Git repositories. Kubernetes does not resolve `secretKeyRef` inside ConfigMap file content. SeaTunnel and Hazelcast must read a final, usable YAML configuration. In production, generate the final ConfigMap through Helm, Kustomize, External Secrets, CI/CD, or the image build process, or use credential mechanisms actually supported by the underlying filesystem, such as Hadoop credential provider, AWS SDK environment variable pr [...] ::: When S3 or OSS is used, the SeaTunnel image must also contain the corresponding Hadoop filesystem implementation and dependencies, such as Hadoop AWS/AWS SDK or Hadoop Aliyun/OSS SDK. Checkpoint storage and IMap MapStore can use the same object storage service, but use different `namespace` prefixes, for example `/seatunnel/checkpoint/` and `/seatunnel/imap`. diff --git a/docs/zh/engines/zeta/checkpoint-storage.md b/docs/zh/engines/zeta/checkpoint-storage.md index 4e87f108a4..b1f58ade27 100644 --- a/docs/zh/engines/zeta/checkpoint-storage.md +++ b/docs/zh/engines/zeta/checkpoint-storage.md @@ -41,6 +41,23 @@ seatunnel: 注意: namespace必须以"/"结尾。 +#### 使用环境变量提供凭据 + +对象存储凭据可以由云 SDK 从环境变量读取。配置对应的凭据 Provider，并且不要在 +`seatunnel.yaml` 中写入 Access Key 或 Secret Key： + +| `storage.type` | 凭据 Provider 配置项 | Provider 类 | 环境变量 | +| --- | --- | --- | --- | +| `s3` | `fs.s3a.aws.credentials.provider` | `com.amazonaws.auth.EnvironmentVariableCredentialsProvider` | `AWS_ACCESS_KEY_ID`、`AWS_SECRET_ACCESS_KEY`，临时凭据可选 `AWS_SESSION_TOKEN` | +| `oss` | `fs.oss.credentials.provider` | `com.aliyun.oss.common.auth.EnvironmentVariableCredentialsProvider` | `OSS_ACCESS_KEY_ID`、`OSS_ACCESS_KEY_SECRET`，临时凭据可选 `OSS_SESSION_TOKEN` | +| `cos` | `fs.cosn.credentials.provider` | `org.apache.hadoop.fs.cosn.auth.EnvironmentVariableCredentialsProvider` | `COSN_SECRET_ID`、`COSN_SECRET_KEY` | + +环境变量名和 Provider 实现由底层 Hadoop 文件系统和云 SDK 固定。请将环境变量注入所有访问 +检查点存储的 SeaTunnel 进程。原有的静态凭据和云平台原生身份认证方式仍然兼容。 + +GCS 不使用上述凭据 Provider 配置项，而是使用 Application Default Credentials。需要密钥文件时 +可设置 `GOOGLE_APPLICATION_CREDENTIALS`，也可以使用 Workload Identity 等 ADC 机制。 + #### OSS 阿里云OSS是基于hdfs-file，所以你可以参考[Hadoop OSS文档](https://hadoop.apache.org/docs/stable/hadoop-aliyun/tools/hadoop-aliyun/index.html)来配置oss. @@ -68,6 +85,24 @@ seatunnel: fs.oss.endpoint: endpoint address ``` +如果要使用环境变量提供凭据，请删除 `fs.oss.accessKeyId` 和 `fs.oss.accessKeySecret`，并选择 +环境变量 Provider。`OSS_ACCESS_KEY_ID` 和 `OSS_ACCESS_KEY_SECRET` 必须对所有 master 和 +worker 可见： + +```yaml +seatunnel: + engine: + checkpoint: + storage: + type: hdfs + plugin-config: + namespace: /seatunnel/checkpoint/ + storage.type: oss + oss.bucket: oss://your-bucket + fs.oss.endpoint: oss-cn-hangzhou.aliyuncs.com + fs.oss.credentials.provider: com.aliyun.oss.common.auth.EnvironmentVariableCredentialsProvider +``` + 有关Hadoop Credential Provider API的更多信息，请参见: [Credential Provider API](https://hadoop.apache.org/docs/stable/hadoop-project-dist/hadoop-common/CredentialProviderAPI.html). 阿里云OSS凭证提供程序实现见: [验证凭证提供](https://github.com/aliyun/aliyun-oss-java-sdk/tree/master/src/main/java/com/aliyun/oss/common/auth) @@ -100,6 +135,17 @@ seatunnel: fs.cosn.bucket.region: your-region ``` +如果要使用环境变量提供凭据，请删除静态的 `secretId` 和 `secretKey`，并配置： + +```yaml + storage.type: cos + cos.bucket: cosn://your-bucket + fs.cosn.bucket.region: your-region + fs.cosn.credentials.provider: org.apache.hadoop.fs.cosn.auth.EnvironmentVariableCredentialsProvider +``` + +`COSN_SECRET_ID` 和 `COSN_SECRET_KEY` 必须对所有 master 和 worker 可见。 + 有关Hadoop Credential Provider API的更多信息，请参见: [Credential Provider API](https://hadoop.apache.org/docs/stable/hadoop-project-dist/hadoop-common/CredentialProviderAPI.html). 腾讯云COS相关配置可参考：[Tencent Hadoop-COS文档](https://doc.fincloud.tencent.cn/tcloud/Storage/COS/846365/hadoop) @@ -186,6 +232,17 @@ seatunnel: ``` +如果要从环境变量读取凭据，请删除 `fs.s3a.access.key` 和 `fs.s3a.secret.key`，并配置： + +```yaml + storage.type: s3 + s3.bucket: s3a://your-bucket + fs.s3a.aws.credentials.provider: com.amazonaws.auth.EnvironmentVariableCredentialsProvider +``` + +`AWS_ACCESS_KEY_ID` 和 `AWS_SECRET_ACCESS_KEY` 必须对所有 master 和 worker 可见；使用临时 +凭据时还需要配置 `AWS_SESSION_TOKEN`。 + 如果您使用`InstanceProfileCredentialsProvider`，它支持在EC2 VM中运行时使用实例配置文件凭据，您可以检查[iam-roles-for-amazon-ec2](https://docs.aws.amazon.com/zh_cn/AWSEC2/latest/UserGuide/iam-roles-for-amazon-ec2.html). 您可以这样配置: diff --git a/docs/zh/engines/zeta/hybrid-cluster-deployment.md b/docs/zh/engines/zeta/hybrid-cluster-deployment.md index 72c1cbeba4..8cac9840a8 100644 --- a/docs/zh/engines/zeta/hybrid-cluster-deployment.md +++ b/docs/zh/engines/zeta/hybrid-cluster-deployment.md @@ -360,14 +360,24 @@ map: fs.oss.endpoint: OSS endpoint ``` +如果要通过环境变量提供 OSS 凭据，请删除 `fs.oss.accessKeyId` 和 +`fs.oss.accessKeySecret`，并增加： + +```yaml + fs.oss.credentials.provider: com.aliyun.oss.common.auth.EnvironmentVariableCredentialsProvider +``` + +该 Provider 读取 `OSS_ACCESS_KEY_ID` 和 `OSS_ACCESS_KEY_SECRET`，临时凭据还可使用 +`OSS_SESSION_TOKEN`。 + 注意：使用OSS 时，确保 lib目录下有这几个jar。 其中 `seatunnel-shade-hadoop3-uber` 来自 [Apache SeaTunnel Shade](https://github.com/apache/seatunnel-shade) 项目，它是对 Hadoop 客户端的 shaded（包重定位）版本，所有第三方类被重定位到 `org.apache.seatunnel.shade.*` 下，避免与 SeaTunnel 自身的依赖产生类路径冲突。版本号格式为 `${library.version}-${seatunnel.shade.version}`（例如 `3.1.4-3.0.0`），具体版本请参考 SeaTunnel 发行包中实际包含的 JAR 文件名。 ``` -aliyun-sdk-oss-3.13.2.jar -hadoop-aliyun-3.3.6.jar -jdom2-2.0.6.jar +aliyun-sdk-oss-3.4.1.jar +hadoop-aliyun-3.1.4.jar +jdom-1.1.jar netty-buffer-4.1.89.Final.jar netty-common-4.1.89.Final.jar seatunnel-shade-hadoop3-uber-${seatunnel.shade.hadoop.version}-${seatunnel.shade.version}.jar diff --git a/docs/zh/engines/zeta/separated-cluster-deployment.md b/docs/zh/engines/zeta/separated-cluster-deployment.md index 2930dacf06..ed0300d5cf 100644 --- a/docs/zh/engines/zeta/separated-cluster-deployment.md +++ b/docs/zh/engines/zeta/separated-cluster-deployment.md @@ -321,14 +321,24 @@ map: fs.oss.endpoint: OSS endpoint ``` +如果要通过环境变量提供 OSS 凭据，请删除 `fs.oss.accessKeyId` 和 +`fs.oss.accessKeySecret`，并增加： + +```yaml + fs.oss.credentials.provider: com.aliyun.oss.common.auth.EnvironmentVariableCredentialsProvider +``` + +该 Provider 读取 `OSS_ACCESS_KEY_ID` 和 `OSS_ACCESS_KEY_SECRET`，临时凭据还可使用 +`OSS_SESSION_TOKEN`。 + 注意：使用OSS 时，确保 lib目录下有这几个jar。 其中 `seatunnel-shade-hadoop3-uber` 来自 [Apache SeaTunnel Shade](https://github.com/apache/seatunnel-shade) 项目，它是对 Hadoop 客户端的 shaded（包重定位）版本，所有第三方类被重定位到 `org.apache.seatunnel.shade.*` 下，避免与 SeaTunnel 自身的依赖产生类路径冲突。版本号格式为 `${library.version}-${seatunnel.shade.version}`（例如 `3.1.4-3.0.0`），具体版本请参考 SeaTunnel 发行包中实际包含的 JAR 文件名。 ``` -aliyun-sdk-oss-3.13.2.jar -hadoop-aliyun-3.3.6.jar -jdom2-2.0.6.jar +aliyun-sdk-oss-3.4.1.jar +hadoop-aliyun-3.1.4.jar +jdom-1.1.jar netty-buffer-4.1.89.Final.jar netty-common-4.1.89.Final.jar seatunnel-shade-hadoop3-uber-${seatunnel.shade.hadoop.version}-${seatunnel.shade.version}.jar diff --git a/docs/zh/engines/zeta/state-storage-and-recovery.md b/docs/zh/engines/zeta/state-storage-and-recovery.md index 5eba3601dd..840dd109f0 100644 --- a/docs/zh/engines/zeta/state-storage-and-recovery.md +++ b/docs/zh/engines/zeta/state-storage-and-recovery.md @@ -58,6 +58,8 @@ seatunnel: # fs.s3a.endpoint: https://s3.amazonaws.com # fs.s3a.access.key: <your-access-key> # fs.s3a.secret.key: <your-secret-key> + # 环境变量凭据： + # fs.s3a.aws.credentials.provider: com.amazonaws.auth.EnvironmentVariableCredentialsProvider ``` ### Checkpoint 保留与清理 diff --git a/docs/zh/getting-started/kubernetes/configuration.md b/docs/zh/getting-started/kubernetes/configuration.md index 426b67994c..35a1e210df 100644 --- a/docs/zh/getting-started/kubernetes/configuration.md +++ b/docs/zh/getting-started/kubernetes/configuration.md @@ -283,7 +283,20 @@ properties: 阿里云 OSS 使用 Hadoop OSS 配置，并通过 `oss.bucket` 指定目标 bucket。`fs.oss.endpoint` 需要按 bucket 所在地域、专有云或兼容 OSS 服务的实际地址填写。 -Hadoop Aliyun OSS 的默认凭据 provider 读取 `fs.oss.accessKeyId`、`fs.oss.accessKeySecret`，临时凭据可额外配置 `fs.oss.securityToken`。它没有像 Hadoop S3A 一样的默认环境变量认证链，因此 Kubernetes Secret 注入成环境变量后不会自动生效。 +Hadoop Aliyun OSS 的默认凭据 provider 读取 `fs.oss.accessKeyId`、`fs.oss.accessKeySecret`，临时凭据可额外配置 `fs.oss.securityToken`。如果要从环境变量读取凭据并且不把凭据渲染到 ConfigMap 中，可以选择 OSS SDK 的环境变量 provider： + +```yaml +properties: + type: hdfs + namespace: /seatunnel/imap + clusterName: seatunnel-cluster + storage.type: oss + oss.bucket: oss://seatunnel-bucket + fs.oss.endpoint: https://oss-<region>.aliyuncs.com + fs.oss.credentials.provider: com.aliyun.oss.common.auth.EnvironmentVariableCredentialsProvider +``` + +该 provider 固定读取环境变量 `OSS_ACCESS_KEY_ID` 和 `OSS_ACCESS_KEY_SECRET`，使用临时凭据时还会读取 `OSS_SESSION_TOKEN`。请将这些变量注入每个 Master 和 Worker Pod。 ```yaml hazelcast: @@ -304,12 +317,59 @@ hazelcast: fs.oss.accessKeySecret: YOUR_ACCESS_KEY_SECRET ``` -如果不希望把 OSS 凭据渲染到最终 ConfigMap，可以使用 Hadoop credential provider 存储 `fs.oss.accessKeyId`、`fs.oss.accessKeySecret`，或通过 `fs.oss.credentials.provider` 指定自定义 provider。自定义 provider 需要实现 Aliyun OSS SDK 的 `com.aliyun.oss.common.auth.CredentialsProvider`，并在 SeaTunnel 镜像中可被加载。 +如果不希望把 OSS 凭据渲染到最终 ConfigMap，可以使用上面的环境变量 provider、Hadoop credential provider，或通过 `fs.oss.credentials.provider` 指定自定义 provider。自定义 provider 需要实现 Aliyun OSS SDK 的 `com.aliyun.oss.common.auth.CredentialsProvider`，并在 SeaTunnel 镜像中可被加载。 + +#### 从 Kubernetes Secret 注入凭据 + +Kubernetes 会在 Pod 启动时将 Secret 的值注入容器进程的环境变量。凭据不会写入 Docker 镜像，也不需要配置 Dockerfile 的 `ENV`。需要为所有可能访问检查点或 IMap 存储的 Master 和 Worker 容器注入凭据。 + +如果 Secret 的 key 已经使用固定的环境变量名，可以直接使用 `envFrom`： + +```yaml +apiVersion: v1 +kind: Secret +metadata: + name: seatunnel-oss-credentials +type: Opaque +stringData: + OSS_ACCESS_KEY_ID: your-access-key + OSS_ACCESS_KEY_SECRET: your-secret-key + OSS_SESSION_TOKEN: your-session-token # 永久凭据不需要这一项 +--- +spec: + template: + spec: + containers: + - name: seatunnel + envFrom: + - secretRef: + name: seatunnel-oss-credentials +``` + +如果 Secret 使用了其他 key 名称，需要显式映射为 Provider 要求的固定变量名： + +```yaml +containers: + - name: seatunnel + env: + - name: OSS_ACCESS_KEY_ID + valueFrom: + secretKeyRef: + name: seatunnel-oss-credentials + key: access-key + - name: OSS_ACCESS_KEY_SECRET + valueFrom: + secretKeyRef: + name: seatunnel-oss-credentials + key: secret-key +``` + +不要给 `envFrom` 设置 `prefix`，因为 Provider 只会读取上面列出的固定变量名。Secret 更新后，运行中容器的环境变量不会自动变化；轮换凭据后需要重启或触发 StatefulSet 滚动更新。 `backup-count: 1` 时，Master 至少需要 2 个副本，才能保存 IMap 备份副本。 :::caution 注意 -不要把真实 access key、secret key 直接提交到 ConfigMap 或 Git 仓库。Kubernetes 不会在 ConfigMap 文件内容中解析 `secretKeyRef`，SeaTunnel/Hazelcast 读取到的必须是最终可用的 YAML 配置。生产环境建议通过 Helm、Kustomize、External Secrets、CI/CD 或镜像构建流程生成最终 ConfigMap，或者使用 Hadoop credential provider、AWS SDK 环境变量 provider、云厂商凭据链、挂载凭据文件等底层文件系统真正支持的凭据方式，并确保所有 Master Pod 具有相同的对象存储读写权限。 +不要把真实 access key、secret key 直接提交到 ConfigMap 或 Git 仓库。Kubernetes 不会在 ConfigMap 文件内容中解析 `secretKeyRef`，SeaTunnel/Hazelcast 读取到的必须是最终可用的 YAML 配置。生产环境建议通过 Helm、Kustomize、External Secrets、CI/CD 或镜像构建流程生成最终 ConfigMap，或者使用 Hadoop credential provider、AWS SDK 环境变量 provider、云厂商凭据链、挂载凭据文件等底层文件系统真正支持的凭据方式，并确保所有 Master 和 Worker Pod 具有相同的对象存储读写权限。 ::: 如果使用 S3 或 OSS，SeaTunnel 镜像中还需要包含对应 Hadoop 文件系统实现及其依赖，例如 Hadoop AWS/AWS SDK 或 Hadoop Aliyun/OSS SDK。Checkpoint 存储和 IMap MapStore 可以使用同一个对象存储服务，但建议使用不同的 `namespace` 前缀，例如 `/seatunnel/checkpoint/` 和 `/seatunnel/imap`。 diff --git a/seatunnel-engine/seatunnel-engine-storage/checkpoint-storage-plugins/checkpoint-storage-hdfs/pom.xml b/seatunnel-engine/seatunnel-engine-storage/checkpoint-storage-plugins/checkpoint-storage-hdfs/pom.xml index 6ff5eb0915..fc437af994 100644 --- a/seatunnel-engine/seatunnel-engine-storage/checkpoint-storage-plugins/checkpoint-storage-hdfs/pom.xml +++ b/seatunnel-engine/seatunnel-engine-storage/checkpoint-storage-plugins/checkpoint-storage-hdfs/pom.xml @@ -31,7 +31,10 @@ <name>SeaTunnel : Engine : Storage : Checkpoint Storage Plugins : HDFS</name> <properties> - <hadoop-aliyun.version>3.0.0</hadoop-aliyun.version> + <hadoop-aliyun.version>3.1.4</hadoop-aliyun.version> + <hadoop-cos.version>3.4.1</hadoop-cos.version> + <aws-java-sdk.version>1.11.271</aws-java-sdk.version> + <aliyun.sdk.oss.version>3.4.1</aliyun.sdk.oss.version> </properties> <dependencies> @@ -50,19 +53,25 @@ <dependency> <groupId>org.apache.hadoop</groupId> <artifactId>hadoop-aws</artifactId> - <version>3.1.4</version> + <version>${hadoop-aws.version}</version> <scope>provided</scope> </dependency> <dependency> <groupId>com.amazonaws</groupId> <artifactId>aws-java-sdk-bundle</artifactId> - <version>1.11.271</version> + <version>${aws-java-sdk.version}</version> <scope>provided</scope> </dependency> <dependency> <groupId>org.apache.hadoop</groupId> <artifactId>hadoop-cos</artifactId> - <version>3.4.1</version> + <version>${hadoop-cos.version}</version> + <scope>provided</scope> + </dependency> + <dependency> + <groupId>com.aliyun.oss</groupId> + <artifactId>aliyun-sdk-oss</artifactId> + <version>${aliyun.sdk.oss.version}</version> <scope>provided</scope> </dependency> </dependencies> diff --git a/seatunnel-engine/seatunnel-engine-storage/checkpoint-storage-plugins/checkpoint-storage-hdfs/src/test/java/org/apache/seatunnel/engine/checkpoint/storage/hdfs/common/EnvironmentCredentialConfigurationTest.java b/seatunnel-engine/seatunnel-engine-storage/checkpoint-storage-plugins/checkpoint-storage-hdfs/src/test/java/org/apache/seatunnel/engine/checkpoint/storage/hdfs/common/EnvironmentCredentialConfigurationTest.java new file mode 100644 index 0000000000..c3cde70e4b --- /dev/null +++ b/seatunnel-engine/seatunnel-engine-storage/checkpoint-storage-plugins/checkpoint-storage-hdfs/src/test/java/org/apache/seatunnel/engine/checkpoint/storage/hdfs/common/EnvironmentCredentialConfigurationTest.java @@ -0,0 +1,220 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.seatunnel.engine.checkpoint.storage.hdfs.common; + +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.fs.aliyun.oss.AliyunOSSUtils; +import org.apache.hadoop.fs.cosn.CosNUtils; +import org.apache.hadoop.fs.s3a.AWSCredentialProviderList; +import org.apache.hadoop.fs.s3a.S3AUtils; + +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.parallel.Execution; +import org.junit.jupiter.api.parallel.ExecutionMode; +import org.junit.jupiter.api.parallel.ResourceLock; + +import com.aliyun.oss.common.auth.Credentials; +import com.aliyun.oss.common.auth.CredentialsProvider; +import com.aliyun.oss.common.auth.EnvironmentVariableCredentialsProvider; +import com.amazonaws.auth.AWSSessionCredentials; +import com.qcloud.cos.auth.COSCredentials; + +import java.net.URI; +import java.util.HashMap; +import java.util.Map; + +import static com.github.stefanbirkner.systemlambda.SystemLambda.withEnvironmentVariable; + +/** + * Tests mutate process-wide environment variables through {@code SystemLambda}. {@code SAME_THREAD} + * keeps the methods in this class sequential, and the shared resource lock keeps other environment- + * mutating test classes from running concurrently. + */ +@Execution(ExecutionMode.SAME_THREAD) +@ResourceLock("system-environment") +public class EnvironmentCredentialConfigurationTest { + + private static final String S3_PROVIDER_KEY = "fs.s3a.aws.credentials.provider"; + private static final String S3_ENV_PROVIDER = + "com.amazonaws.auth.EnvironmentVariableCredentialsProvider"; + + private static final String OSS_PROVIDER_KEY = "fs.oss.credentials.provider"; + private static final String OSS_ENV_PROVIDER = + "com.aliyun.oss.common.auth.EnvironmentVariableCredentialsProvider"; + + private static final String COS_PROVIDER_KEY = "fs.cosn.credentials.provider"; + private static final String COS_ENV_PROVIDER = + "org.apache.hadoop.fs.cosn.auth.EnvironmentVariableCredentialsProvider"; + + private static final String S3_ACCESS_KEY = "test-s3-access-key"; + private static final String S3_SECRET_KEY = "test-s3-secret-key"; + private static final String S3_SESSION_TOKEN = "test-s3-session-token"; + + private static final String OSS_ACCESS_KEY = "test-oss-access-key"; + private static final String OSS_SECRET_KEY = "test-oss-secret-key"; + private static final String OSS_SESSION_TOKEN = "test-oss-session-token"; + + private static final String COS_SECRET_ID = "test-cos-secret-id"; + private static final String COS_SECRET_KEY = "test-cos-secret-key"; + + @Test + public void testS3EnvironmentCredentialProviderIsPassedThrough() throws Exception { + Map<String, String> config = new HashMap<>(); + config.put("s3.bucket", "s3a://seatunnel-checkpoint"); + config.put(S3_PROVIDER_KEY, S3_ENV_PROVIDER); + + Configuration hadoopConf = + FileConfiguration.valueOf("S3").getConfiguration().buildConfiguration(config); + + Assertions.assertEquals(S3_ENV_PROVIDER, hadoopConf.get(S3_PROVIDER_KEY)); + Assertions.assertNull(hadoopConf.get("fs.s3a.access.key")); + Assertions.assertNull(hadoopConf.get("fs.s3a.secret.key")); + } + + @Test + public void testOssEnvironmentCredentialProviderIsPassedThrough() throws Exception { + Map<String, String> config = new HashMap<>(); + config.put("oss.bucket", "oss://seatunnel-checkpoint"); + config.put("fs.oss.endpoint", "oss-cn-hangzhou.aliyuncs.com"); + config.put(OSS_PROVIDER_KEY, OSS_ENV_PROVIDER); + + Configuration hadoopConf = + FileConfiguration.valueOf("OSS").getConfiguration().buildConfiguration(config); + + Assertions.assertEquals(OSS_ENV_PROVIDER, hadoopConf.get(OSS_PROVIDER_KEY)); + Assertions.assertNull(hadoopConf.get("fs.oss.accessKeyId")); + Assertions.assertNull(hadoopConf.get("fs.oss.accessKeySecret")); + } + + @Test + public void testCosEnvironmentCredentialProviderIsPassedThrough() throws Exception { + Map<String, String> config = new HashMap<>(); + config.put("cos.bucket", "cosn://seatunnel-checkpoint"); + config.put("fs.cosn.bucket.region", "ap-shanghai"); + config.put(COS_PROVIDER_KEY, COS_ENV_PROVIDER); + + Configuration hadoopConf = + FileConfiguration.valueOf("COS").getConfiguration().buildConfiguration(config); + + Assertions.assertEquals(COS_ENV_PROVIDER, hadoopConf.get(COS_PROVIDER_KEY)); + Assertions.assertNull(hadoopConf.get("fs.cosn.userinfo.secretId")); + Assertions.assertNull(hadoopConf.get("fs.cosn.userinfo.secretKey")); + } + + /** + * Guards that the provider classes are present on the test classpath; the tests below verify + * that they actually read the environment variables. + */ + @Test + public void testEnvironmentCredentialProviderClassesAreOnTestClasspath() { + Assertions.assertDoesNotThrow(() -> Class.forName(S3_ENV_PROVIDER)); + Assertions.assertDoesNotThrow(() -> Class.forName(OSS_ENV_PROVIDER)); + Assertions.assertDoesNotThrow(() -> Class.forName(COS_ENV_PROVIDER)); + } + + @Test + public void testS3EnvironmentProviderReadsConfiguredValues() throws Exception { + withEnvironmentVariable("AWS_ACCESS_KEY_ID", S3_ACCESS_KEY) + .and("AWS_SECRET_ACCESS_KEY", S3_SECRET_KEY) + .and("AWS_SESSION_TOKEN", S3_SESSION_TOKEN) + .execute( + () -> { + Map<String, String> config = new HashMap<>(); + config.put("s3.bucket", "s3a://seatunnel-checkpoint"); + config.put(S3_PROVIDER_KEY, S3_ENV_PROVIDER); + Configuration hadoopConf = + FileConfiguration.valueOf("S3") + .getConfiguration() + .buildConfiguration(config); + + try (AWSCredentialProviderList providers = + S3AUtils.createAWSCredentialProviderSet( + URI.create("s3a://seatunnel-checkpoint"), hadoopConf)) { + AWSSessionCredentials credentials = + Assertions.assertInstanceOf( + AWSSessionCredentials.class, + providers.getCredentials()); + + Assertions.assertEquals( + S3_ACCESS_KEY, credentials.getAWSAccessKeyId()); + Assertions.assertEquals( + S3_SECRET_KEY, credentials.getAWSSecretKey()); + Assertions.assertEquals( + S3_SESSION_TOKEN, credentials.getSessionToken()); + } + }); + } + + @Test + public void testOssEnvironmentProviderReadsConfiguredValues() throws Exception { + withEnvironmentVariable("OSS_ACCESS_KEY_ID", OSS_ACCESS_KEY) + .and("OSS_ACCESS_KEY_SECRET", OSS_SECRET_KEY) + .and("OSS_SESSION_TOKEN", OSS_SESSION_TOKEN) + .execute( + () -> { + Map<String, String> config = new HashMap<>(); + config.put("oss.bucket", "oss://seatunnel-checkpoint"); + config.put("fs.oss.endpoint", "oss-cn-hangzhou.aliyuncs.com"); + config.put(OSS_PROVIDER_KEY, OSS_ENV_PROVIDER); + Configuration hadoopConf = + FileConfiguration.valueOf("OSS") + .getConfiguration() + .buildConfiguration(config); + + CredentialsProvider provider = + AliyunOSSUtils.getCredentialsProvider( + URI.create("oss://seatunnel-checkpoint"), hadoopConf); + Assertions.assertInstanceOf( + EnvironmentVariableCredentialsProvider.class, provider); + Credentials credentials = provider.getCredentials(); + + Assertions.assertEquals(OSS_ACCESS_KEY, credentials.getAccessKeyId()); + Assertions.assertEquals( + OSS_SECRET_KEY, credentials.getSecretAccessKey()); + Assertions.assertEquals( + OSS_SESSION_TOKEN, credentials.getSecurityToken()); + }); + } + + @Test + public void testCosEnvironmentProviderReadsConfiguredValues() throws Exception { + withEnvironmentVariable("COSN_SECRET_ID", COS_SECRET_ID) + .and("COSN_SECRET_KEY", COS_SECRET_KEY) + .execute( + () -> { + Map<String, String> config = new HashMap<>(); + config.put("cos.bucket", "cosn://seatunnel-checkpoint"); + config.put("fs.cosn.bucket.region", "ap-shanghai"); + config.put(COS_PROVIDER_KEY, COS_ENV_PROVIDER); + Configuration hadoopConf = + FileConfiguration.valueOf("COS") + .getConfiguration() + .buildConfiguration(config); + + COSCredentials credentials = + CosNUtils.createCosCredentialsProviderSet( + URI.create("cosn://seatunnel-checkpoint"), + hadoopConf) + .getCredentials(); + + Assertions.assertEquals(COS_SECRET_ID, credentials.getCOSAccessKeyId()); + Assertions.assertEquals(COS_SECRET_KEY, credentials.getCOSSecretKey()); + }); + } +} diff --git a/seatunnel-engine/seatunnel-engine-storage/imap-storage-plugins/imap-storage-file/pom.xml b/seatunnel-engine/seatunnel-engine-storage/imap-storage-plugins/imap-storage-file/pom.xml index cb7bf09e26..cea7b72349 100644 --- a/seatunnel-engine/seatunnel-engine-storage/imap-storage-plugins/imap-storage-file/pom.xml +++ b/seatunnel-engine/seatunnel-engine-storage/imap-storage-plugins/imap-storage-file/pom.xml @@ -32,7 +32,8 @@ <properties> <!-- Imap storage dependency package --> - <hadoop-aliyun.version>3.0.0</hadoop-aliyun.version> + <hadoop-aliyun.version>3.1.4</hadoop-aliyun.version> + <aliyun.sdk.oss.version>3.4.1</aliyun.sdk.oss.version> <json-smart.version>2.4.7</json-smart.version> <netty-buffer.version>4.1.60.Final</netty-buffer.version> </properties> @@ -91,28 +92,30 @@ </exclusion> </exclusions> </dependency> - <dependency> <groupId>net.minidev</groupId> <artifactId>json-smart</artifactId> <version>${json-smart.version}</version> <scope>provided</scope> </dependency> - <dependency> <groupId>org.apache.hadoop</groupId> <artifactId>hadoop-aws</artifactId> <version>${hadoop-aws.version}</version> <scope>provided</scope> </dependency> - <dependency> <groupId>io.netty</groupId> <artifactId>netty-buffer</artifactId> <version>${netty-buffer.version}</version> <scope>provided</scope> </dependency> - + <dependency> + <groupId>com.aliyun.oss</groupId> + <artifactId>aliyun-sdk-oss</artifactId> + <version>${aliyun.sdk.oss.version}</version> + <scope>provided</scope> + </dependency> </dependencies> </project> diff --git a/seatunnel-engine/seatunnel-engine-storage/imap-storage-plugins/imap-storage-file/src/test/java/org/apache/seatunnel/engine/imap/storage/file/config/EnvironmentCredentialConfigurationTest.java b/seatunnel-engine/seatunnel-engine-storage/imap-storage-plugins/imap-storage-file/src/test/java/org/apache/seatunnel/engine/imap/storage/file/config/EnvironmentCredentialConfigurationTest.java new file mode 100644 index 0000000000..a22260100d --- /dev/null +++ b/seatunnel-engine/seatunnel-engine-storage/imap-storage-plugins/imap-storage-file/src/test/java/org/apache/seatunnel/engine/imap/storage/file/config/EnvironmentCredentialConfigurationTest.java @@ -0,0 +1,130 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package org.apache.seatunnel.engine.imap.storage.file.config; + +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.fs.aliyun.oss.AliyunOSSUtils; +import org.apache.hadoop.fs.s3a.AWSCredentialProviderList; +import org.apache.hadoop.fs.s3a.S3AUtils; + +import org.junit.jupiter.api.Assertions; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.parallel.Execution; +import org.junit.jupiter.api.parallel.ExecutionMode; +import org.junit.jupiter.api.parallel.ResourceLock; + +import com.aliyun.oss.common.auth.Credentials; +import com.aliyun.oss.common.auth.CredentialsProvider; +import com.aliyun.oss.common.auth.EnvironmentVariableCredentialsProvider; +import com.amazonaws.auth.AWSSessionCredentials; + +import java.net.URI; +import java.util.HashMap; +import java.util.Map; + +import static com.github.stefanbirkner.systemlambda.SystemLambda.withEnvironmentVariable; + +/** + * Tests mutate process-wide environment variables through {@code SystemLambda}. {@code SAME_THREAD} + * keeps the methods in this class sequential, and the shared resource lock keeps other environment- + * mutating test classes from running concurrently. + */ +@Execution(ExecutionMode.SAME_THREAD) +@ResourceLock("system-environment") +public class EnvironmentCredentialConfigurationTest { + + private static final String S3_PROVIDER_KEY = "fs.s3a.aws.credentials.provider"; + private static final String S3_ENV_PROVIDER = + "com.amazonaws.auth.EnvironmentVariableCredentialsProvider"; + + private static final String OSS_PROVIDER_KEY = "fs.oss.credentials.provider"; + private static final String OSS_ENV_PROVIDER = + "com.aliyun.oss.common.auth.EnvironmentVariableCredentialsProvider"; + + private static final String S3_ACCESS_KEY = "imap-s3-access-key"; + private static final String S3_SECRET_KEY = "imap-s3-secret-key"; + private static final String S3_SESSION_TOKEN = "imap-s3-session-token"; + + private static final String OSS_ACCESS_KEY = "imap-oss-access-key"; + private static final String OSS_SECRET_KEY = "imap-oss-secret-key"; + private static final String OSS_SESSION_TOKEN = "imap-oss-session-token"; + + @Test + public void testS3EnvironmentProviderReadsConfiguredValues() throws Exception { + withEnvironmentVariable("AWS_ACCESS_KEY_ID", S3_ACCESS_KEY) + .and("AWS_SECRET_ACCESS_KEY", S3_SECRET_KEY) + .and("AWS_SESSION_TOKEN", S3_SESSION_TOKEN) + .execute( + () -> { + Map<String, String> config = new HashMap<>(); + config.put("s3.bucket", "s3a://seatunnel-imap"); + config.put(S3_PROVIDER_KEY, S3_ENV_PROVIDER); + Configuration hadoopConf = + FileConfiguration.valueOf("S3") + .getConfiguration() + .buildConfiguration(config); + + try (AWSCredentialProviderList providers = + S3AUtils.createAWSCredentialProviderSet( + URI.create("s3a://seatunnel-imap"), hadoopConf)) { + AWSSessionCredentials credentials = + Assertions.assertInstanceOf( + AWSSessionCredentials.class, + providers.getCredentials()); + + Assertions.assertEquals( + S3_ACCESS_KEY, credentials.getAWSAccessKeyId()); + Assertions.assertEquals( + S3_SECRET_KEY, credentials.getAWSSecretKey()); + Assertions.assertEquals( + S3_SESSION_TOKEN, credentials.getSessionToken()); + } + }); + } + + @Test + public void testOssEnvironmentProviderReadsConfiguredValues() throws Exception { + withEnvironmentVariable("OSS_ACCESS_KEY_ID", OSS_ACCESS_KEY) + .and("OSS_ACCESS_KEY_SECRET", OSS_SECRET_KEY) + .and("OSS_SESSION_TOKEN", OSS_SESSION_TOKEN) + .execute( + () -> { + Map<String, String> config = new HashMap<>(); + config.put("oss.bucket", "oss://seatunnel-imap"); + config.put("fs.oss.endpoint", "oss-cn-hangzhou.aliyuncs.com"); + config.put(OSS_PROVIDER_KEY, OSS_ENV_PROVIDER); + Configuration hadoopConf = + FileConfiguration.valueOf("OSS") + .getConfiguration() + .buildConfiguration(config); + + CredentialsProvider provider = + AliyunOSSUtils.getCredentialsProvider( + URI.create("oss://seatunnel-imap"), hadoopConf); + Assertions.assertInstanceOf( + EnvironmentVariableCredentialsProvider.class, provider); + Credentials credentials = provider.getCredentials(); + + Assertions.assertEquals(OSS_ACCESS_KEY, credentials.getAccessKeyId()); + Assertions.assertEquals( + OSS_SECRET_KEY, credentials.getSecretAccessKey()); + Assertions.assertEquals( + OSS_SESSION_TOKEN, credentials.getSecurityToken()); + }); + } +}
