Copilot commented on code in PR #7473: URL: https://github.com/apache/texera/pull/7473#discussion_r3744347157
########## 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: `None` does not necessarily mean “shared warehouse”: `VFSURIFactory.decodeURI` also returns `None` for an invalid leading `/wh/<name>` prefix. This therefore allows an unresolvable warehouse URI through, after which `DocumentFactory` opens the shared catalog—the silent fallback that #6930 explicitly requires rejecting. Preserve whether a `/wh/` prefix was present and fail when it cannot be resolved. ########## 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: This guard covers only websocket result pagination. Other user-facing reads still call `DocumentFactory.openDocument` directly; for example, `GET /executions/{wid}/stats/{eid}` does so in `WorkflowExecutionsResource.scala:723-724`. Consequently, runtime statistics in a per-user warehouse remain readable while the kill switch is off. Enforce the gate at a shared URI-opening boundary or apply it to every read path. ########## 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: This endpoint is persisted in each Lakekeeper warehouse profile, but local development deliberately uses a host/LAN endpoint that can change between starts. The existing default-warehouse initializer refreshes that endpoint on every run for exactly this reason (`bin/single-node/docker-compose.yml:221-256`); per-user warehouses have no equivalent refresh and will become unusable after the address changes. Refresh all Local warehouse profiles at startup or use a stable reachable endpoint. ########## 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: Persisting `whid` alone does not enable the stated last-used-warehouse preselection. Both execution-list queries and `retrieveLatestExecutionEntry` select an explicit field list without `WHID`, and `WorkflowExecutionEntry` has no warehouse field (`WorkflowExecutionsResource.scala:318-340, 522-535, 580-606`). Expose the stored value through the latest-execution response, as is already done for `CUID`. ########## 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: When the flag is off and no warehouse is selected, this resolves the new run to the shared warehouse, but the following cleanup still opens and clears the previous execution’s stored URIs. If that previous run used a per-user warehouse, starting a default run while the kill switch is off reaches into and deletes data from that disabled warehouse. The cleanup must skip or explicitly reject warehouse-scoped URIs while disabled. ########## 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: If this DB write or the subsequent refresh fails after Lakekeeper creation succeeds, the request fails while leaving an untracked Lakekeeper warehouse that the user cannot list or delete through Texera. Wrap persistence in a compensating cleanup (with logging/reconciliation if cleanup also fails) so create cannot orphan remote state. -- 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]
