mengw15 commented on code in PR #7473: URL: https://github.com/apache/texera/pull/7473#discussion_r3744415635
########## amber/src/main/scala/org/apache/texera/web/service/WarehouseReadGuard.scala: ########## @@ -0,0 +1,43 @@ +/* + * 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.texera.web.service + +import org.apache.texera.common.config.StorageConfig + +/** + * Guards reads of results that live in a per-user warehouse while the feature is off (#6930). + * + * The warehouse switch is a kill switch: turning it off must disable reads too, and it must + * fail *explicitly*. Without this guard a `/wh/<name>/…` URI would resolve to the shared + * default warehouse and surface "table not found" — indistinguishable from data loss. No data + * is lost; re-enabling the switch restores access. + */ +object WarehouseReadGuard { + + def assertReadable( + warehouse: Option[String], + enabled: Boolean = StorageConfig.warehouseEnabled + ): Unit = + warehouse.filterNot(_ => enabled).foreach { name => Review Comment: Done in e10e0831f — the guard now takes the URI and refuses a `/wh/` prefix that decodes to no warehouse, in either flag state. ########## amber/src/main/scala/org/apache/texera/web/service/ExecutionResultService.scala: ########## @@ -481,6 +481,11 @@ class ExecutionResultService( PortIdentity() ) + // Refuse to read a per-user-warehouse result while the feature is off (#6930). + storageUriOption.foreach(uri => + WarehouseReadGuard.assertReadable(VFSURIFactory.decodeURI(uri).warehouse) + ) Review Comment: Done in e10e0831f — the guard now also covers the stats read, result export, and the sync API's result/console reads; cleanup paths skip warehouse URIs instead (see the WorkflowService thread). ########## amber/src/main/scala/org/apache/texera/web/service/WorkflowService.scala: ########## @@ -198,6 +233,7 @@ class WorkflowService( ) val workflowContext: WorkflowContext = createWorkflowContext() + workflowContext.warehouse = WorkflowService.resolveWarehouseName(req.warehouseId, uid) Review Comment: Done in e10e0831f — every cleanup path (new-run clearing, execution deletion, workflow deletion) now skips warehouse-scoped URIs while the feature is off, with a log line. ########## amber/src/main/scala/org/apache/texera/web/service/ExecutionsMetadataPersistService.scala: ########## @@ -71,6 +72,8 @@ object ExecutionsMetadataPersistService extends LazyLogging { // Set computing unit ID if provided newExecution.setCuid(computingUnitId) + // The warehouse this run writes into (#6870); null = the shared default warehouse. + warehouseId.foreach(whid => newExecution.setWhid(whid)) Review Comment: Done in e10e0831f — `WorkflowExecutionEntry` gains `whId` and both entry queries select `WHID`, mirroring `cuId`; pinned in WorkflowExecutionsResourceSpec. ########## amber/src/main/scala/org/apache/texera/web/resource/dashboard/user/warehouse/WarehouseResource.scala: ########## @@ -0,0 +1,187 @@ +/* + * 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.texera.web.resource.dashboard.user.warehouse + +import io.dropwizard.auth.Auth +import org.apache.texera.auth.SessionUser +import org.apache.texera.common.config.StorageConfig +import org.apache.texera.dao.SqlServer +import org.apache.texera.dao.jooq.generated.Tables.USER_WAREHOUSE +import org.apache.texera.dao.jooq.generated.enums.UserWarehouseFlavorEnum +import org.apache.texera.web.resource.dashboard.user.warehouse.WarehouseResource._ +import org.apache.texera.web.service.LakekeeperClient + +import javax.annotation.security.RolesAllowed +import javax.ws.rs._ +import javax.ws.rs.core.MediaType + +object WarehouseResource { + private def context = + SqlServer + .getInstance() + .createDSLContext() + + // A warehouse's user-facing name becomes part of the Lakekeeper catalog name + // `user-<uid>-<name>`, which in turn becomes a VFS URI path segment — so it is + // restricted to the same characters VFSURIFactory accepts for a warehouse name. + private val warehouseNamePattern = "[A-Za-z0-9][A-Za-z0-9_-]*".r + + private[warehouse] def isValidWarehouseName(name: String): Boolean = + name.length <= 64 && warehouseNamePattern.pattern.matcher(name).matches() + + case class DashboardWarehouse( + whid: Integer, + name: String, + warehouseName: String, + flavor: String, + createdAtMillis: Long + ) + + case class WarehouseStatus(enabled: Boolean, warehouses: List[DashboardWarehouse]) + + case class CreateWarehouseRequest(name: String) +} + +/** + * Per-user warehouse management (#6870): list the feature state and the caller's + * warehouses, create a Local-flavor warehouse on the deployment's own object store, + * and delete one (empty-first in Lakekeeper, purging its data files). + * + * Everything except `/status` is gated by [[StorageConfig.warehouseEnabled]]; the + * mutating endpoints return 403 while the flag is off. `/status` always answers so the + * frontend can decide whether to show the feature at all. + */ +@Path("/warehouse") +@Produces(Array(MediaType.APPLICATION_JSON)) +class WarehouseResource(client: LakekeeperClient) { + + def this() = this(new LakekeeperClient()) + + @GET + @Path("/status") + def status(@Auth current_user: SessionUser): WarehouseStatus = { + if (!StorageConfig.warehouseEnabled) { + return WarehouseStatus(enabled = false, warehouses = List()) + } + val warehouses = context + .selectFrom(USER_WAREHOUSE) + .where(USER_WAREHOUSE.UID.eq(current_user.getUid)) + .orderBy(USER_WAREHOUSE.CREATED_AT.asc()) + .fetch() + .map(row => + DashboardWarehouse( + row.getWhid, + row.getName, + row.getWarehouseName, + row.getFlavor.getLiteral, + row.getCreatedAt.toInstant.toEpochMilli + ) + ) + WarehouseStatus( + enabled = true, + warehouses = warehouses.toArray(Array[DashboardWarehouse]()).toList + ) + } + + @POST + @RolesAllowed(Array("REGULAR", "ADMIN")) + def create( + request: CreateWarehouseRequest, + @Auth current_user: SessionUser + ): DashboardWarehouse = { + requireEnabled() + val name = Option(request.name).map(_.trim).getOrElse("") + if (!isValidWarehouseName(name)) { + throw new BadRequestException( + "warehouse name must start with a letter or digit and contain only letters, " + + "digits, '-' and '_' (at most 64 characters)" + ) + } + val uid = current_user.getUid + if ( + context.fetchExists( + context + .selectFrom(USER_WAREHOUSE) + .where(USER_WAREHOUSE.UID.eq(uid).and(USER_WAREHOUSE.NAME.eq(name))) + ) + ) { + throw new WebApplicationException(s"a warehouse named '$name' already exists", 409) + } + + val warehouseName = s"user-$uid-$name" + // Create in Lakekeeper first, record after: a failed creation leaves no orphaned row. + val warehouseId = + try { + client.createWarehouse(warehouseName) + } catch { + case e: Exception => + throw new WebApplicationException(e.getMessage, 502) + } + + val row = context.newRecord(USER_WAREHOUSE) + row.setUid(uid) + row.setName(name) + row.setWarehouseName(warehouseName) + row.setLakekeeperWarehouseId(warehouseId) + row.setFlavor(UserWarehouseFlavorEnum.local) + row.setS3Bucket(StorageConfig.icebergRESTCatalogS3Bucket) + row.setS3Endpoint(StorageConfig.s3Endpoint) + row.setS3Region(StorageConfig.s3Region) + row.store() + // created_at is filled by the DB default; fetch it back before serializing. + row.refresh() Review Comment: Done in e10e0831f — a failed row write now compensates by deleting the just-created (empty) warehouse, logging if the compensation itself fails. ########## amber/src/main/scala/org/apache/texera/web/service/LakekeeperClient.scala: ########## @@ -0,0 +1,151 @@ +/* + * 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.texera.web.service + +import com.fasterxml.jackson.databind.ObjectMapper +import kong.unirest.Unirest +import org.apache.texera.common.config.StorageConfig + +import java.net.URLEncoder +import java.nio.charset.StandardCharsets +import java.util.UUID +import scala.jdk.CollectionConverters.IteratorHasAsScala + +/** + * Client for the Lakekeeper APIs used to manage per-user warehouses (#6870). + * + * Two API families are involved: the **management** API (`/management/v1/...`) creates and + * deletes warehouse entities, and the **catalog** API (`/catalog/v1/{warehouseId}/...`) lists + * and drops the namespaces/tables inside one. The channel is unauthenticated today; + * catalog-side authentication is Phase 2 (#6040). + * + * @param catalogUri the Iceberg REST catalog uri (ends with `/catalog`), from which the + * management base is derived. Overridable for tests. + */ +class LakekeeperClient(catalogUri: String = StorageConfig.icebergRESTCatalogUri) { + + // Lakekeeper's default project; single-project deployments (ours) use the nil UUID. + private val DefaultProjectId = "00000000-0000-0000-0000-000000000000" + + private val managementBase: String = catalogUri.stripSuffix("/catalog") + "/management/v1" + private val catalogBase: String = catalogUri + "/v1" + + private val mapper = new ObjectMapper() + + private def urlEncode(segment: String): String = + URLEncoder.encode(segment, StandardCharsets.UTF_8) + + private def failOn(status: Int, body: String, action: String): Unit = + if (status < 200 || status >= 300) { + throw new RuntimeException(s"Lakekeeper $action failed (HTTP $status): $body") + } + + /** + * Creates a warehouse backed by this deployment's own object store (the Local flavor): + * the storage profile points at the configured MinIO/S3 endpoint and bucket, with the + * platform's static credentials and STS off. + * + * @return the Lakekeeper-assigned warehouse id. + */ + def createWarehouse(warehouseName: String): UUID = { + val payload = mapper.createObjectNode() + payload.put("warehouse-name", warehouseName) + payload.put("project-id", DefaultProjectId) + + val profile = payload.putObject("storage-profile") + profile.put("type", "s3") + profile.put("bucket", StorageConfig.icebergRESTCatalogS3Bucket) + profile.put("region", StorageConfig.s3Region) + profile.put("endpoint", StorageConfig.s3Endpoint) Review Comment: Right — this is the same failure mode as #7358 (the default warehouse's endpoint going stale in local dev). Production uses a stable in-cluster DNS name, so this bites local dev only; extending #7358's startup refresh to every Local warehouse covers it, so I've widened that issue's scope rather than growing this PR. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
