<p>I'm trying to deploy a Flink stream processor to a Kubernetes cluster with the help of the official <a href="https://nightlies.apache.org/flink/flink-kubernetes-operator-docs-stable/" rel="nofollow noreferrer">Flink kubernetes operator</a>.
The Flink app also uses Minio as its state backend. Everything worked fine until I tried to provide the credentials from Hashicorp Vault in the following way:</p>
<pre class="lang-yaml prettyprint-override"><code>apiVersion: flink.apache.org/v1beta1
kind: FlinkDeployment
metadata:
  name: flink-app
  namespace: default
spec:
  serviceAccount: sa-example
  podTemplate:
    apiVersion: v1
    kind: Pod
    metadata:
      name: pod-template
    spec:
      serviceAccountName: default:sa-example
      containers:
        - name: flink-main-container
      # ....
  flinkVersion: v1_14
  flinkConfiguration:
    presto.s3.endpoint: https://s3-example-api.dev.net
    high-availability: org.apache.flink.kubernetes.highavailability.KubernetesHaServicesFactory
    high-availability.storageDir: s3p://example-flink/example-1/high-availability/
    high-availability.cluster-id: example-1
    high-availability.namespace: example
    high-availability.service-account: default:sa-example
#    presto.s3.access-key: *
#    presto.s3.secret-key: *  
    presto.s3.path-style-access: &quot;true&quot;
    web.upload.dir: /opt/flink
  jobManager:
    podTemplate:
      apiVersion: v1
      kind: Pod
      metadata:
        name: job-manager-pod-template
        annotations:
          vault.hashicorp.com/namespace: &quot;/example/dev&quot;
          vault.hashicorp.com/agent-inject: &quot;true&quot;
          vault.hashicorp.com/agent-init-first: &quot;true&quot;
          vault.hashicorp.com/agent-inject-secret-appsecrets.yaml: &quot;example/Minio&quot;
          vault.hashicorp.com/role: &quot;example-serviceaccount&quot;
          vault.hashicorp.com/auth-path: auth/example
          vault.hashicorp.com/agent-inject-template-appsecrets.yaml: |
            {{- with secret &quot;example/Minio&quot; -}}
            presto.s3.access-key: {{.Data.data.accessKey}}
            presto.s3.secret-key: {{.Data.data.secretKey}}
            {{- end }}
</code></pre>
<p>When I comment the <code>presto.s3.access-key</code> and <code>presto.s3.secret-key</code> config values in the flinkConfiguration, replace them with the above listed <code>Hashicorp Vault</code> annotations and try to provide them programmatically during runtime:</p>
<pre class="lang-scala prettyprint-override"><code>val configuration: Configuration = getSecretsFromFile(&quot;/vault/secrets/appsecrets.yaml&quot;)

val env = org.apache.flink.streaming.api.environment.StreamExecutionEnvironment.getExecutionEnvironment(configuration)
</code></pre>
<p>I receive the following error message:</p>
<p><em>java.io.IOException: com.amazonaws.SdkClientException: Unable to load AWS credentials from any provider in the chain: [EnvironmentVariableCredentialsProvider: Unable to load AWS credentials from environment variables (AWS_ACCESS_KEY_ID (or AWS_ACCESS_KEY) and AWS_SECRET_KEY (or AWS_SECRET_ACCESS_KEY)), SystemPropertiesCredentialsProvider: Unable to load AWS credentials from Java system properties (aws.accessKeyId and aws.secretKey), WebIdentityTokenCredentialsProvider: You must specify a value for roleArn and roleSessionName, com.amazonaws.auth.profile.ProfileCredentialsProvider@5331f738: profile file cannot be null, com.amazonaws.auth.EC2ContainerCredentialsProviderWrapper@bc0353f: Failed to connect to service endpoint: ]
at com.facebook.presto.hive.s3.PrestoS3FileSystem$PrestoS3OutputStream.uploadObject(PrestoS3FileSystem.java:1278) ~[flink-s3-fs-presto-1.14.2.jar:1.14.2]
at com.facebook.presto.hive.s3.PrestoS3FileSystem$PrestoS3OutputStream.close(PrestoS3FileSystem.java:1226) ~[flink-s3-fs-presto-1.14.2.jar:1.14.2]
at org.apache.hadoop.fs.FSDataOutputStream$PositionCache.close(FSDataOutputStream.java:72) ~[flink-s3-fs-presto-1.14.2.jar:1.14.2]
at org.apache.hadoop.fs.FSDataOutputStream.close(FSDataOutputStream.java:101) ~[flink-s3-fs-presto-1.14.2.jar:1.14.2]
at org.apache.flink.fs.s3presto.common.HadoopDataOutputStream.close(HadoopDataOutputStream.java:52) ~[flink-s3-fs-presto-1.14.2.jar:1.14.2]
at org.apache.flink.runtime.blob.FileSystemBlobStore.put(FileSystemBlobStore.java:80) ~[flink-dist_2.12-1.14.2.jar:1.14.2]
at org.apache.flink.runtime.blob.FileSystemBlobStore.put(FileSystemBlobStore.java:72) ~[flink-dist_2.12-1.14.2.jar:1.14.2]
at org.apache.flink.runtime.blob.BlobUtils.moveTempFileToStore(BlobUtils.java:385) ~[flink-dist_2.12-1.14.2.jar:1.14.2]
at org.apache.flink.runtime.blob.BlobServer.moveTempFileToStore(BlobServer.java:680) ~[flink-dist_2.12-1.14.2.jar:1.14.2]
at org.apache.flink.runtime.blob.BlobServerConnection.put(BlobServerConnection.java:350) [flink-dist_2.12-1.14.2.jar:1.14.2]
at org.apache.flink.runtime.blob.BlobServerConnection.run(BlobServerConnection.java:110) [flink-dist_2.12-1.14.2.jar:1.14.2]</em></p>
<p>I initially also tried to append the secrets to flink-config.yaml in the docker-entrypoint.sh based on this <a href="https://nightlies.apache.org/flink/flink-docs-release-1.15/docs/deployment/filesystems/s3/#configure-access-credentials" rel="nofollow noreferrer">documentation - Configure Access Credentials</a>:</p>
<pre><code>if [ -f '/vault/secrets/appsecrets.yaml' ]; then
  (echo &amp;&amp; cat '/vault/secrets/appsecrets.yaml') &gt;&gt; $FLINK_HOME/conf/flink-conf.yaml
fi
</code></pre>
<p>The question is how to provide the S3 credentials during the runtime since the Flink operator mounts the <strong>flink-config.yaml</strong> from a config map and it is a flink-conf.yaml: <strong>Read-only file system</strong>.</p>
<p>Thank you</p>

Kubus
<p>There is no support for this from the Kubernetes operator. In fact, this is not a limitation of the Flink Kubernetes operator, it is due to the fact of lack in support in Kubernetes native integration. There is a separate story for this in the Kubernetes operator side - <a href="https://issues.apache.org/jira/browse/FLINK-27491" rel="nofollow noreferrer">FLINK-27491</a>.</p>
<p>As a workaround, what you can do is, set up an init container and update the config map from the init container using kubernetes API after reading it from the vault. So the updated config map should have the secrets replaced by the init container and those will be visible to the job manager and all of its task managers. The whole Flink cluster journey starts only after updating the config map from the init container so it should be visible to the Flink cluster.</p>
<p>A simple example to update the config map from the init container can be found <a href="https://stackoverflow.com/questions/52046908/are-kubernetes-configmaps-writable#answer-71662405">here</a>. In this example, the config map is updated with a simple CURL command. In theory, you can use any lightweight client to update the config map like <a href="https://kubernetes.io/docs/tasks/run-application/access-api-from-pod/" rel="nofollow noreferrer">this</a>.</p>
<p>A side note: If possible I would suggest to use AWS IAM role rather than IAM plain secrets as IAM role is more secure compared to IAM static credentials.</p>
