Support STS for OSS ufs through RAMRole

core/common/src/main/java/alluxio/conf/PropertyKey.java

@@ -1275,6 +1275,29 @@ public String toString() {
           .setConsistencyCheckLevel(ConsistencyCheckLevel.WARN)
           .setScope(Scope.SERVER)
           .build();
+  public static final PropertyKey UNDERFS_OSS_STS_ENABLED =
+      booleanBuilder(Name.UNDERFS_OSS_STS_ENABLED)
+          .setAlias("alluxio.underfs.oss.sts.enabled")
+          .setDefaultValue(false)
+          .setDescription("Whether to enable oss STS(Security Token Service).")
+          .setConsistencyCheckLevel(ConsistencyCheckLevel.WARN)
+          .setScope(Scope.SERVER)
+          .build();
+  public static final PropertyKey UNDERFS_OSS_RETRY_MAX =
+      intBuilder(Name.UNDERFS_OSS_RETRY_MAX)
+          .setAlias("alluxio.underfs.oss.retry.max")
+          .setDefaultValue(3)
+          .setDescription("The maximum number of OSS error retry.")
+          .setConsistencyCheckLevel(ConsistencyCheckLevel.WARN)
+          .setScope(Scope.SERVER)
+          .build();
+  public static final PropertyKey UNDERFS_OSS_ECS_RAM_ROLE =
+      stringBuilder(Name.UNDERFS_OSS_ECS_RAM_ROLE)
+          .setAlias("alluxio.underfs.oss.ecs.ram.role")
+          .setDescription("The RAM role of current owner of ECS.")
+          .setConsistencyCheckLevel(ConsistencyCheckLevel.WARN)
+          .setScope(Scope.SERVER)
+          .build();
   public static final PropertyKey UNDERFS_S3_ADMIN_THREADS_MAX =
       intBuilder(Name.UNDERFS_S3_ADMIN_THREADS_MAX)
           .setDefaultValue(20)
@@ -7238,6 +7261,9 @@ public static final class Name {
         "alluxio.underfs.oss.connection.timeout";
     public static final String UNDERFS_OSS_CONNECT_TTL = "alluxio.underfs.oss.connection.ttl";
     public static final String UNDERFS_OSS_SOCKET_TIMEOUT = "alluxio.underfs.oss.socket.timeout";
+    public static final String UNDERFS_OSS_STS_ENABLED = "alluxio.underfs.oss.sts.enabled";
+    public static final String UNDERFS_OSS_RETRY_MAX = "alluxio.underfs.oss.retry.max";
+    public static final String UNDERFS_OSS_ECS_RAM_ROLE = "alluxio.underfs.oss.ecs.ram.role";
     public static final String UNDERFS_S3_BULK_DELETE_ENABLED =
         "alluxio.underfs.s3.bulk.delete.enabled";
     public static final String UNDERFS_S3_DEFAULT_MODE = "alluxio.underfs.s3.default.mode";

underfs/oss/src/main/java/alluxio/underfs/oss/OSSUnderFileSystem.java

@@ -59,6 +59,9 @@ public class OSSUnderFileSystem extends ObjectUnderFileSystem {
   /** Bucket name of user's configured Alluxio bucket. */
   private final String mBucketName;
 
+  private boolean mStsEnabled;
+  private StsOssClientProvider mClientProvider;
+
   /**
    * Constructs a new instance of {@link OSSUnderFileSystem}.
    *
@@ -69,18 +72,22 @@ public class OSSUnderFileSystem extends ObjectUnderFileSystem {
   public static OSSUnderFileSystem createInstance(AlluxioURI uri, UnderFileSystemConfiguration conf)
       throws Exception {
     String bucketName = UnderFileSystemUtils.getBucketName(uri);
-    Preconditions.checkArgument(conf.isSet(PropertyKey.OSS_ACCESS_KEY),
-        "Property %s is required to connect to OSS", PropertyKey.OSS_ACCESS_KEY);
-    Preconditions.checkArgument(conf.isSet(PropertyKey.OSS_SECRET_KEY),
-        "Property %s is required to connect to OSS", PropertyKey.OSS_SECRET_KEY);
-    Preconditions.checkArgument(conf.isSet(PropertyKey.OSS_ENDPOINT_KEY),
-        "Property %s is required to connect to OSS", PropertyKey.OSS_ENDPOINT_KEY);
-    String accessId = conf.getString(PropertyKey.OSS_ACCESS_KEY);
-    String accessKey = conf.getString(PropertyKey.OSS_SECRET_KEY);
-    String endPoint = conf.getString(PropertyKey.OSS_ENDPOINT_KEY);
-
-    ClientBuilderConfiguration ossClientConf = initializeOSSClientConfig(conf);
-    OSS ossClient = new OSSClientBuilder().build(endPoint, accessId, accessKey, ossClientConf);
+
+    OSS ossClient = null;
+    if (!conf.getBoolean(PropertyKey.UNDERFS_OSS_STS_ENABLED)) {
+      Preconditions.checkArgument(conf.isSet(PropertyKey.OSS_ACCESS_KEY),
+              "Property %s is required to connect to OSS", PropertyKey.OSS_ACCESS_KEY);
+      Preconditions.checkArgument(conf.isSet(PropertyKey.OSS_SECRET_KEY),
+              "Property %s is required to connect to OSS", PropertyKey.OSS_SECRET_KEY);
+      Preconditions.checkArgument(conf.isSet(PropertyKey.OSS_ENDPOINT_KEY),
+              "Property %s is required to connect to OSS", PropertyKey.OSS_ENDPOINT_KEY);
+      String accessId = conf.getString(PropertyKey.OSS_ACCESS_KEY);
+      String accessKey = conf.getString(PropertyKey.OSS_SECRET_KEY);
+      String endPoint = conf.getString(PropertyKey.OSS_ENDPOINT_KEY);
+
+      ClientBuilderConfiguration ossClientConf = initializeOSSClientConfig(conf);
+      ossClient = new OSSClientBuilder().build(endPoint, accessId, accessKey, ossClientConf);
+    }
 
     return new OSSUnderFileSystem(uri, ossClient, bucketName, conf);
   }
@@ -96,7 +103,18 @@ public static OSSUnderFileSystem createInstance(AlluxioURI uri, UnderFileSystemC
   protected OSSUnderFileSystem(AlluxioURI uri, OSS ossClient, String bucketName,
       UnderFileSystemConfiguration conf) {
     super(uri, conf);
-    mClient = ossClient;
+
+    mStsEnabled = conf.getBoolean(PropertyKey.UNDERFS_OSS_STS_ENABLED);
+    if (mStsEnabled) {
+      try {
+        mClientProvider = new StsOssClientProvider(conf);
+      } catch (IOException e) {
+        LOG.error("init sts client provider failed!", e);
+        throw new ServiceException(e);
+      }
+    }
+    mClient = mStsEnabled ? mClientProvider.getOSSClient() : ossClient;
+
     mBucketName = bucketName;
   }
 

underfs/oss/src/main/java/alluxio/underfs/oss/OSSUnderFileSystemFactory.java

@@ -62,6 +62,10 @@ public boolean supportsPath(String path) {
    * @return true if both access, secret and endpoint keys are present, false otherwise
    */
   private boolean checkOSSCredentials(UnderFileSystemConfiguration conf) {
+    if (conf.getBoolean(PropertyKey.UNDERFS_OSS_STS_ENABLED)) {
+      return conf.isSet(PropertyKey.UNDERFS_OSS_ECS_RAM_ROLE);
+    }
+
     return conf.isSet(PropertyKey.OSS_ACCESS_KEY)
         && conf.isSet(PropertyKey.OSS_SECRET_KEY)
         && conf.isSet(PropertyKey.OSS_ENDPOINT_KEY);

underfs/oss/src/main/java/alluxio/underfs/oss/StsOssClientProvider.java

@@ -0,0 +1,238 @@
+/*
+ * The Alluxio Open Foundation licenses this work under the Apache License, version 2.0
+ * (the "License"). You may not use this work except in compliance with the License, which is
+ * available at www.apache.org/licenses/LICENSE-2.0
+ *
+ * This software is distributed on an "AS IS" basis, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND,
+ * either express or implied, as more fully set forth in the License.
+ *
+ * See the NOTICE file distributed with this work for information regarding copyright ownership.
+ */
+
+package alluxio.underfs.oss;
+
+import alluxio.conf.PropertyKey;
+import alluxio.retry.ExponentialBackoffRetry;
+import alluxio.retry.RetryPolicy;
+import alluxio.underfs.UnderFileSystemConfiguration;
+import alluxio.util.ThreadFactoryUtils;
+import alluxio.util.network.HttpUtils;
+
+import com.aliyun.oss.ClientBuilderConfiguration;
+import com.aliyun.oss.OSS;
+import com.aliyun.oss.OSSClientBuilder;
+import com.aliyun.oss.common.auth.DefaultCredentials;
+import com.google.gson.Gson;
+import com.google.gson.JsonObject;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.text.ParseException;
+import java.text.SimpleDateFormat;
+import java.util.Date;
+import java.util.TimeZone;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * STS client provider for Aliyun OSS.
+ */
+public class StsOssClientProvider implements Closeable {
+  private static final Logger LOG = LoggerFactory.getLogger(StsOssClientProvider.class);
+
+  private static final int ECS_META_GET_TIMEOUT = 10000;
+  private static final int BASE_SLEEP_TIME_MS = 1000;
+  private static final int MAX_SLEEP_MS = 3000;
+  private static final int MAX_RETRIES = 5;
+
+  private volatile OSS mOssClient = null;
+  private Date mStsTokenExpiration = null;
+
+  public static final String ECS_METADATA_SERVICE =
+      "http://100.100.100.200/latest/meta-data/ram/security-credentials/";
+
+  private static final int IN_TOKEN_EXPIRED_MS = 1800000;
+  private static final String ACCESS_KEY_ID = "AccessKeyId";
+  private static final String ACCESS_KEY_SECRET = "AccessKeySecret";
+  private static final String SECURITY_TOKEN = "SecurityToken";
+  private static final String EXPIRATION = "Expiration";
+
+  private UnderFileSystemConfiguration mOssConf;
+  private final ScheduledExecutorService mRefreshOssClientScheduledThread;
+
+  /**
+   * Constructs a new instance of {@link StsOssClientProvider}.
+   * @param ossConfiguration {@link UnderFileSystemConfiguration} for OSS
+   * @throws IOException if failed to init OSS STS client
+   */
+  public StsOssClientProvider(UnderFileSystemConfiguration ossConfiguration) throws IOException {
+    RetryPolicy retryPolicy = new ExponentialBackoffRetry(
+        BASE_SLEEP_TIME_MS, MAX_SLEEP_MS, MAX_RETRIES);
+    IOException lastException = null;
+    while (retryPolicy.attempt()) {
+      try {
+        initializeOssClient(ossConfiguration);
+        lastException = null;
+        break;
+      } catch (IOException e) {
+        LOG.warn("init oss client failed! has retry {} times", retryPolicy.getAttemptCount(), e);
+        lastException = e;
+      }
+    }
+    if (lastException != null) {
+      throw lastException;
+    }
+    mRefreshOssClientScheduledThread = Executors.newSingleThreadScheduledExecutor(
+        ThreadFactoryUtils.build("refresh_oss_client-%d", false));
+    mRefreshOssClientScheduledThread.scheduleAtFixedRate(() -> {
+      try {
+        if (null != mOssConf) {
+          refreshOssStsClient(mOssConf);
+        }
+      } catch (Exception e) {
+        //retry it
+        LOG.warn("throw exception when clear meta data cache", e);
+      }
+    }, 0, 60000, TimeUnit.MILLISECONDS);
+  }
+
+  protected void refreshOssStsClient(UnderFileSystemConfiguration ossConfiguration)
+      throws IOException {
+    ClientBuilderConfiguration ossClientConf = getClientBuilderConfiguration(ossConfiguration);
+    createOrRefreshStsOssClient(ossConfiguration, ossClientConf);
+  }
+
+  private ClientBuilderConfiguration getClientBuilderConfiguration(
+      UnderFileSystemConfiguration ossConfiguration) {
+    ClientBuilderConfiguration ossClientConf = new ClientBuilderConfiguration();
+    ossClientConf.setMaxConnections(ossConfiguration.getInt(PropertyKey.UNDERFS_OSS_CONNECT_MAX));
+    ossClientConf.setMaxErrorRetry(ossConfiguration.getInt(PropertyKey.UNDERFS_OSS_RETRY_MAX));
+    ossClientConf.setConnectionTimeout(
+        (int) ossConfiguration.getMs(PropertyKey.UNDERFS_OSS_CONNECT_TIMEOUT));
+    ossClientConf.setSocketTimeout(
+        (int) ossConfiguration.getMs(PropertyKey.UNDERFS_OSS_SOCKET_TIMEOUT));
+    ossClientConf.setSupportCname(false);
+    ossClientConf.setCrcCheckEnabled(true);
+    return ossClientConf;
+  }
+
+  /**
+   * Init the STS OSS client.
+   * @param ossConfiguration OSS {@link UnderFileSystemConfiguration}
+   * @throws IOException if failed to init OSS client
+   */
+  public void initializeOssClient(UnderFileSystemConfiguration ossConfiguration)
+      throws IOException {
+    mOssConf = ossConfiguration;
+    if (null != mOssClient) {
+      return;
+    }
+
+    ClientBuilderConfiguration clientConf = getClientBuilderConfiguration(ossConfiguration);
+    createOrRefreshStsOssClient(ossConfiguration, clientConf);
+
+    LOG.info("init ossClient success : {}", mOssClient.toString());
+  }
+
+  boolean isStsTokenExpired() {
+    boolean expired = true;
+    Date now = convertLongToDate(System.currentTimeMillis());
+    if (null != mStsTokenExpiration) {
+      if (mStsTokenExpiration.after(now)) {
+        expired = false;
+      }
+    }
+    return expired;
+  }
+
+  boolean isTokenWillExpired() {
+    boolean in = true;
+    Date now = convertLongToDate(System.currentTimeMillis());
+    long millisecond = mStsTokenExpiration.getTime() - now.getTime();
+    if (millisecond >= IN_TOKEN_EXPIRED_MS) {
+      in = false;
+    }
+    return in;
+  }
+
+  private void createOrRefreshStsOssClient(
+      UnderFileSystemConfiguration ossConfiguration,
+      ClientBuilderConfiguration clientConfiguration) throws IOException {
+    if (isStsTokenExpired() || isTokenWillExpired()) {
+      try {
+        String ecsRamRole = ossConfiguration.getString(PropertyKey.UNDERFS_OSS_ECS_RAM_ROLE);
+        String fullECSMetaDataServiceUrl = ECS_METADATA_SERVICE + ecsRamRole;
+        String jsonStringResponse = HttpUtils.get(fullECSMetaDataServiceUrl, ECS_META_GET_TIMEOUT);
+
+        JsonObject jsonObject = new Gson().fromJson(jsonStringResponse, JsonObject.class);
+        String accessKeyId = jsonObject.get(ACCESS_KEY_ID).getAsString();
+        String accessKeySecret = jsonObject.get(ACCESS_KEY_SECRET).getAsString();
+        String securityToken = jsonObject.get(SECURITY_TOKEN).getAsString();
+        mStsTokenExpiration = convertStringToDate(jsonObject.get(EXPIRATION).getAsString());
+
+        if (null == mOssClient) {
+          mOssClient = new OSSClientBuilder().build(
+              ossConfiguration.getString(PropertyKey.OSS_ENDPOINT_KEY),
+              accessKeyId, accessKeySecret, securityToken,
+              clientConfiguration);
+        } else {
+          mOssClient.switchCredentials((new DefaultCredentials(
+              accessKeyId, accessKeySecret, securityToken)));
+        }
+        LOG.info("oss sts client create success {} {} {}",
+            mOssClient, securityToken, mStsTokenExpiration);
+      } catch (IOException e) {
+        LOG.error("create stsOssClient exception", e);
+        throw new IOException("create stsOssClient exception", e);
+      }
+    }
+  }
+
+  /**
+   * Returns the STS OSS client.
+   * @return oss client
+   */
+  public OSS getOSSClient() {
+    return mOssClient;
+  }
+
+  private Date convertLongToDate(long timeMs) {
+    TimeZone zeroTimeZone = TimeZone.getTimeZone("ETC/GMT-0");
+    SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss'Z'");
+    sdf.setTimeZone(zeroTimeZone);
+    Date date = null;
+    try {
+      date = sdf.parse(sdf.format(new Date(timeMs)));
+    } catch (ParseException e) {
+      LOG.error("convert String to Date type error", e);
+    }
+    return date;
+  }
+
+  private Date convertStringToDate(String dateString) {
+    TimeZone zeroTimeZone = TimeZone.getTimeZone("ETC/GMT-0");
+    SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd'T'HH:mm:ss'Z'");
+    sdf.setTimeZone(zeroTimeZone);
+    Date date = null;
+    try {
+      date = sdf.parse(dateString);
+    } catch (ParseException e) {
+      LOG.error("convert String to Date type error", e);
+    }
+    return date;
+  }
+
+  @Override
+  public void close() throws IOException {
+    if (null != mRefreshOssClientScheduledThread) {
+      mRefreshOssClientScheduledThread.shutdown();
+    }
+    if (null != mOssClient) {
+      mOssClient.shutdown();
+      mOssClient = null;
+    }
+  }
+}