[openhouse] Create GET /accesss endpoint for fetching DataAccessCredentials for a given table

cluster/storage/src/main/java/com/linkedin/openhouse/cluster/storage/BaseStorage.java

@@ -1,6 +1,7 @@
 package com.linkedin.openhouse.cluster.storage;
 
 import com.google.common.base.Preconditions;
+import com.linkedin.openhouse.cluster.storage.auth.DataAccessCredential;
 import com.linkedin.openhouse.cluster.storage.configs.StorageProperties;
 import java.net.URI;
 import java.util.HashMap;
@@ -76,4 +77,11 @@ public String allocateTableLocation(
         .normalize()
         .toString();
   }
+
+  /** Default implementation returns Optional.empty. */
+  @Override
+  public Optional<DataAccessCredential> getDataAccessCredentialForTableLocation(
+      String tableLocation, Map<String, String> params) {
+    return Optional.empty();
+  }
 }

cluster/storage/src/main/java/com/linkedin/openhouse/cluster/storage/Storage.java

@@ -1,6 +1,8 @@
 package com.linkedin.openhouse.cluster.storage;
 
+import com.linkedin.openhouse.cluster.storage.auth.DataAccessCredential;
 import java.util.Map;
+import java.util.Optional;
 
 /**
  * The Storage interface represents a storage system in OpenHouse. It provides methods to check if
@@ -70,4 +72,15 @@ public interface Storage {
    */
   String allocateTableLocation(
       String databaseId, String tableId, String tableUUID, String tableCreator);
+
+  /**
+   * Returns an optional DataAccessCredential for the given table location.
+   *
+   * @param tableLocation the table location
+   * @param params input parameters needed to get a data access credential for the given table
+   *     location
+   * @return DataAccessCredential for the given table location if possible. Otherwise, empty.
+   */
+  Optional<DataAccessCredential> getDataAccessCredentialForTableLocation(
+      String tableLocation, Map<String, String> params);
 }

cluster/storage/src/main/java/com/linkedin/openhouse/cluster/storage/auth/DataAccessCredential.java

@@ -0,0 +1,22 @@
+package com.linkedin.openhouse.cluster.storage.auth;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+import io.swagger.v3.oas.annotations.media.Schema;
+import java.util.Map;
+import lombok.Builder;
+import lombok.Value;
+
+@Value
+@Builder
+public class DataAccessCredential {
+
+  @Schema(
+      description = "Map with the access credentials",
+      example = "{'token':'header.payload.signature', 'path':'/my/table'}")
+  @JsonProperty(access = JsonProperty.Access.READ_ONLY)
+  private Map<String, String> credential;
+
+  @Schema(description = "Expiration date of token in millis since epoch", example = "0")
+  @JsonProperty(access = JsonProperty.Access.READ_ONLY)
+  private long expirationMillisSinceEpoch;
+}

integrations/java/openhouse-java-runtime/src/main/java/com/linkedin/openhouse/javaclient/OpenHouseTableOperations.java

@@ -92,6 +92,7 @@ public void doRefresh() {
                 WebClientRequestException.class,
                 e -> Mono.error(new WebClientRequestWithMessageException(e)))
             .blockOptional();
+
     if (!tableLocation.isPresent() && currentMetadataLocation() != null) {
       throw new NoSuchTableException(
           "Cannot find table %s after refresh, maybe another process deleted it", tableName());

services/common/src/main/java/com/linkedin/openhouse/common/exception/UnsupportedClientOperationException.java

@@ -18,6 +18,7 @@ public enum Operation {
     ALTER_RESERVED_TBLPROPS,
     ALTER_RESERVED_ROLES,
     GRANT_ON_UNSHARED_TABLES,
-    ALTER_TABLE_TYPE
+    ALTER_TABLE_TYPE,
+    DATA_ACCESS_CREDENTIAL_UNSUPPORTED
   }
 }

services/tables/src/main/java/com/linkedin/openhouse/tables/api/handler/TablesApiHandler.java

@@ -5,7 +5,9 @@
 import com.linkedin.openhouse.tables.api.spec.v0.request.UpdateAclPoliciesRequestBody;
 import com.linkedin.openhouse.tables.api.spec.v0.response.GetAclPoliciesResponseBody;
 import com.linkedin.openhouse.tables.api.spec.v0.response.GetAllTablesResponseBody;
+import com.linkedin.openhouse.tables.api.spec.v0.response.GetDataAccessCredentialResponseBody;
 import com.linkedin.openhouse.tables.api.spec.v0.response.GetTableResponseBody;
+import java.util.Map;
 
 /**
  * Interface layer between REST and Tables backend. The implementation is injected into the Service
@@ -108,4 +110,17 @@ ApiResponse<GetAclPoliciesResponseBody> getAclPolicies(
    */
   ApiResponse<GetAclPoliciesResponseBody> getAclPoliciesForUserPrincipal(
       String databaseId, String tableId, String actingPrincipal, String userPrincipal);
+
+  /**
+   * Function to get data access credential for a Table Resource identified by tableId in a given
+   * databaseId.
+   *
+   * @param databaseId
+   * @param tableId
+   * @param params
+   * @return an access token to read data for the tableId if supported by the table's underlying
+   *     data storage.
+   */
+  ApiResponse<GetDataAccessCredentialResponseBody> getDataAccessCredential(
+      String databaseId, String tableId, Map<String, String> params);
 }

services/tables/src/main/java/com/linkedin/openhouse/tables/api/handler/impl/OpenHouseTablesApiHandler.java

@@ -1,17 +1,23 @@
 package com.linkedin.openhouse.tables.api.handler.impl;
 
 import com.linkedin.openhouse.cluster.configs.ClusterProperties;
+import com.linkedin.openhouse.cluster.storage.auth.DataAccessCredential;
 import com.linkedin.openhouse.common.api.spec.ApiResponse;
+import com.linkedin.openhouse.common.exception.UnsupportedClientOperationException;
 import com.linkedin.openhouse.tables.api.handler.TablesApiHandler;
 import com.linkedin.openhouse.tables.api.spec.v0.request.CreateUpdateTableRequestBody;
 import com.linkedin.openhouse.tables.api.spec.v0.request.UpdateAclPoliciesRequestBody;
 import com.linkedin.openhouse.tables.api.spec.v0.response.GetAclPoliciesResponseBody;
 import com.linkedin.openhouse.tables.api.spec.v0.response.GetAllTablesResponseBody;
+import com.linkedin.openhouse.tables.api.spec.v0.response.GetDataAccessCredentialResponseBody;
 import com.linkedin.openhouse.tables.api.spec.v0.response.GetTableResponseBody;
 import com.linkedin.openhouse.tables.api.validator.TablesApiValidator;
 import com.linkedin.openhouse.tables.dto.mapper.TablesMapper;
 import com.linkedin.openhouse.tables.model.TableDto;
 import com.linkedin.openhouse.tables.services.TablesService;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Optional;
 import java.util.stream.Collectors;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.data.util.Pair;
@@ -24,6 +30,14 @@
 @Component
 public class OpenHouseTablesApiHandler implements TablesApiHandler {
 
+  // Default DataAccessCredential that is returned when no credential is generated for a given
+  // table.
+  private static final DataAccessCredential UNSUPPORTED_DATA_ACCESS_CREDENTIAL =
+      DataAccessCredential.builder()
+          .credential(new HashMap<>())
+          .expirationMillisSinceEpoch(-1)
+          .build();
+
   @Autowired private TablesApiValidator tablesApiValidator;
 
   @Autowired private TablesService tableService;
@@ -140,4 +154,27 @@ public ApiResponse<GetAclPoliciesResponseBody> getAclPoliciesForUserPrincipal(
                 .build())
         .build();
   }
+
+  @Override
+  public ApiResponse<GetDataAccessCredentialResponseBody> getDataAccessCredential(
+      String databaseId, String tableId, Map<String, String> params) {
+    Optional<DataAccessCredential> dataAccessCredential =
+        tableService.getDataAccessCredential(databaseId, tableId, params);
+
+    if (!dataAccessCredential.isPresent()) {
+      throw new UnsupportedClientOperationException(
+          UnsupportedClientOperationException.Operation.DATA_ACCESS_CREDENTIAL_UNSUPPORTED,
+          "Unable to get a DataAccessCredential for the given table.");
+    }
+
+    return ApiResponse.<GetDataAccessCredentialResponseBody>builder()
+        .httpStatus(HttpStatus.OK)
+        .responseBody(
+            GetDataAccessCredentialResponseBody.builder()
+                .credential(dataAccessCredential.get().getCredential())
+                .expirationMillisSinceEpoch(
+                    dataAccessCredential.get().getExpirationMillisSinceEpoch())
+                .build())
+        .build();
+  }
 }

services/tables/src/main/java/com/linkedin/openhouse/tables/api/spec/v0/response/GetDataAccessCredentialResponseBody.java

@@ -0,0 +1,27 @@
+package com.linkedin.openhouse.tables.api.spec.v0.response;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.google.gson.Gson;
+import io.swagger.v3.oas.annotations.media.Schema;
+import java.util.Map;
+import lombok.Builder;
+import lombok.Value;
+
+@Builder(toBuilder = true)
+@Value
+public class GetDataAccessCredentialResponseBody {
+
+  @Schema(
+      description = "Map with the access credentials",
+      example = "{'token':'header.payload.signature', 'path':'/my/table'}")
+  @JsonProperty(access = JsonProperty.Access.READ_ONLY)
+  private Map<String, String> credential;
+
+  @Schema(description = "Expiration date of token in millis since epoch", example = "0")
+  @JsonProperty(access = JsonProperty.Access.READ_ONLY)
+  private long expirationMillisSinceEpoch;
+
+  public String toJson() {
+    return new Gson().toJson(this);
+  }
+}

services/tables/src/main/java/com/linkedin/openhouse/tables/controller/TablesController.java

@@ -7,13 +7,16 @@
 import com.linkedin.openhouse.tables.api.spec.v0.request.UpdateAclPoliciesRequestBody;
 import com.linkedin.openhouse.tables.api.spec.v0.response.GetAclPoliciesResponseBody;
 import com.linkedin.openhouse.tables.api.spec.v0.response.GetAllTablesResponseBody;
+import com.linkedin.openhouse.tables.api.spec.v0.response.GetDataAccessCredentialResponseBody;
 import com.linkedin.openhouse.tables.api.spec.v0.response.GetTableResponseBody;
 import com.linkedin.openhouse.tables.authorization.Privileges;
 import io.swagger.v3.oas.annotations.Operation;
 import io.swagger.v3.oas.annotations.Parameter;
 import io.swagger.v3.oas.annotations.media.Schema;
 import io.swagger.v3.oas.annotations.responses.ApiResponse;
 import io.swagger.v3.oas.annotations.responses.ApiResponses;
+import java.util.Collections;
+import java.util.Map;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.http.ResponseEntity;
 import org.springframework.security.access.annotation.Secured;
@@ -291,4 +294,39 @@ public ResponseEntity<GetAclPoliciesResponseBody> getAclPoliciesForUserPrincipal
     return new ResponseEntity<>(
         apiResponse.getResponseBody(), apiResponse.getHttpHeaders(), apiResponse.getHttpStatus());
   }
+
+  @Operation(
+      summary = "Get temporary credentials to access data in a given table",
+      description =
+          "Returns the temporary credentials which have data access to the Table resource identified by by "
+              + "databaseId and tableId. The expiration time of the temporary credentials is returned "
+              + "in millis since epoch.",
+      tags = {"Table"})
+  @ApiResponses(
+      value = {
+        @ApiResponse(responseCode = "200", description = "Access GET: OK"),
+        @ApiResponse(responseCode = "400", description = "Access GET: ACCESS_UNSUPPORTED"),
+      })
+  @GetMapping(
+      value = {
+        "/v0/databases/{databaseId}/tables/{tableId}/access",
+        "/v1/databases/{databaseId}/tables/{tableId}/access"
+      },
+      produces = {"application/json"})
+  public ResponseEntity<GetDataAccessCredentialResponseBody> getDataAccessCredential(
+      @Parameter(description = "Database ID", required = true) @PathVariable String databaseId,
+      @Parameter(description = "Table ID", required = true) @PathVariable String tableId,
+      @Parameter(description = "Other Params", required = false) @PathVariable
+          Map<String, String> params) {
+
+    if (params == null) {
+      params = Collections.emptyMap();
+    }
+
+    com.linkedin.openhouse.common.api.spec.ApiResponse<GetDataAccessCredentialResponseBody>
+        apiResponse = tablesApiHandler.getDataAccessCredential(databaseId, tableId, params);
+
+    return new ResponseEntity<>(
+        apiResponse.getResponseBody(), apiResponse.getHttpHeaders(), apiResponse.getHttpStatus());
+  }
 }

services/tables/src/main/java/com/linkedin/openhouse/tables/services/TablesService.java

@@ -1,10 +1,13 @@
 package com.linkedin.openhouse.tables.services;
 
+import com.linkedin.openhouse.cluster.storage.auth.DataAccessCredential;
 import com.linkedin.openhouse.tables.api.spec.v0.request.CreateUpdateTableRequestBody;
 import com.linkedin.openhouse.tables.api.spec.v0.request.UpdateAclPoliciesRequestBody;
 import com.linkedin.openhouse.tables.api.spec.v0.response.components.AclPolicy;
 import com.linkedin.openhouse.tables.model.TableDto;
 import java.util.List;
+import java.util.Map;
+import java.util.Optional;
 import org.springframework.data.util.Pair;
 
 /** Service Interface for Implementing /tables endpoint. */
@@ -93,4 +96,8 @@ void updateAclPolicies(
    */
   List<AclPolicy> getAclPolicies(
       String databaseId, String tableId, String actingPrincipal, String userPrincipal);
+
+  /** Get DataAccessCredential for the given table */
+  Optional<DataAccessCredential> getDataAccessCredential(
+      String databaseId, String tableId, Map<String, String> params);
 }

services/tables/src/main/java/com/linkedin/openhouse/tables/services/TablesServiceImpl.java

@@ -1,5 +1,9 @@
 package com.linkedin.openhouse.tables.services;
 
+import com.linkedin.openhouse.cluster.storage.Storage;
+import com.linkedin.openhouse.cluster.storage.StorageManager;
+import com.linkedin.openhouse.cluster.storage.StorageType;
+import com.linkedin.openhouse.cluster.storage.auth.DataAccessCredential;
 import com.linkedin.openhouse.common.api.spec.TableUri;
 import com.linkedin.openhouse.common.exception.AlreadyExistsException;
 import com.linkedin.openhouse.common.exception.EntityConcurrentModificationException;
@@ -18,6 +22,7 @@
 import com.linkedin.openhouse.tables.repository.OpenHouseInternalRepository;
 import com.linkedin.openhouse.tables.utils.AuthorizationUtils;
 import com.linkedin.openhouse.tables.utils.TableUUIDGenerator;
+import java.net.URI;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
@@ -41,6 +46,9 @@ public class TablesServiceImpl implements TablesService {
   @Autowired AuthorizationHandler authorizationHandler;
 
   @Autowired TableUUIDGenerator tableUUIDGenerator;
+
+  @Autowired StorageManager storageManager;
+
   /**
    * Lookup a table by databaseId and tableId in OpenHouse's Internal Catalog.
    *
@@ -211,6 +219,18 @@ public List<AclPolicy> getAclPolicies(
     return authorizationHandler.listAclPolicies(tableDto, userPrincipal);
   }
 
+  @Override
+  public Optional<DataAccessCredential> getDataAccessCredential(
+      String databaseId, String tableId, Map<String, String> params) {
+    TableDto tableDto = getTableOrThrow(databaseId, tableId);
+    StorageType.Type tableStorageType =
+        new StorageType().fromString(URI.create(tableDto.getTableLocation()).getScheme());
+
+    Storage storage = storageManager.getStorage(tableStorageType);
+
+    return storage.getDataAccessCredentialForTableLocation(tableDto.getTableLocation(), params);
+  }
+
   /** Whether sharing has been enabled for the table denoted by tableDto. */
   private boolean isTableSharingEnabled(TableDto tableDto) {
     return (tableDto.getPolicies() != null && tableDto.getPolicies().isSharingEnabled());

services/tables/src/test/java/com/linkedin/openhouse/tables/mock/MockTablesApiHandler.java

@@ -14,7 +14,9 @@
 import com.linkedin.openhouse.tables.api.spec.v0.request.UpdateAclPoliciesRequestBody;
 import com.linkedin.openhouse.tables.api.spec.v0.response.GetAclPoliciesResponseBody;
 import com.linkedin.openhouse.tables.api.spec.v0.response.GetAllTablesResponseBody;
+import com.linkedin.openhouse.tables.api.spec.v0.response.GetDataAccessCredentialResponseBody;
 import com.linkedin.openhouse.tables.api.spec.v0.response.GetTableResponseBody;
+import java.util.Map;
 import org.springframework.context.annotation.Primary;
 import org.springframework.http.HttpStatus;
 import org.springframework.security.access.AccessDeniedException;
@@ -187,6 +189,23 @@ public ApiResponse<GetAclPoliciesResponseBody> getAclPoliciesForUserPrincipal(
     }
   }
 
+  @Override
+  public ApiResponse<GetDataAccessCredentialResponseBody> getDataAccessCredential(
+      String databaseId, String tableId, Map<String, String> params) {
+    switch (databaseId) {
+      case "d200":
+        return ApiResponse.<GetDataAccessCredentialResponseBody>builder()
+            .httpStatus(HttpStatus.OK)
+            .responseBody(RequestConstants.TEST_GET_DATA_ACCESS_CREDENTIAL_RESPONSE_BODY)
+            .build();
+      case "d400":
+        throw new UnsupportedClientOperationException(
+            UnsupportedClientOperationException.Operation.DATA_ACCESS_CREDENTIAL_UNSUPPORTED, "");
+      default:
+        throw new RuntimeException("Unknown databaseId: " + databaseId);
+    }
+  }
+
   private void throwTableException(String tableId) {
     switch (tableId) {
       case "entityconcurrentmodificationexception":

services/tables/src/test/java/com/linkedin/openhouse/tables/mock/RequestConstants.java

@@ -6,6 +6,7 @@
 import com.linkedin.openhouse.tables.api.spec.v0.response.GetAclPoliciesResponseBody;
 import com.linkedin.openhouse.tables.api.spec.v0.response.GetAllDatabasesResponseBody;
 import com.linkedin.openhouse.tables.api.spec.v0.response.GetAllTablesResponseBody;
+import com.linkedin.openhouse.tables.api.spec.v0.response.GetDataAccessCredentialResponseBody;
 import com.linkedin.openhouse.tables.api.spec.v0.response.GetDatabaseResponseBody;
 import com.linkedin.openhouse.tables.api.spec.v0.response.GetTableResponseBody;
 import com.linkedin.openhouse.tables.api.spec.v0.response.components.AclPolicy;
@@ -114,4 +115,11 @@ private RequestConstants() {}
       AclPolicy.builder().principal("TEST_USER").role("TABLE_ADMIN").build();
   public static final GetAclPoliciesResponseBody TEST_GET_ACL_POLICIES_RESPONSE_BODY =
       GetAclPoliciesResponseBody.builder().results(Collections.singletonList(ACL_POLICY)).build();
+
+  public static final GetDataAccessCredentialResponseBody
+      TEST_GET_DATA_ACCESS_CREDENTIAL_RESPONSE_BODY =
+          GetDataAccessCredentialResponseBody.builder()
+              .credential(Collections.singletonMap("token", "header.payload.signature"))
+              .expirationMillisSinceEpoch(0)
+              .build();
 }