Offloader: add API to scan objects on Tiered Storage

(cherry picked from commit 154b450)

managed-ledger/src/main/java/org/apache/bookkeeper/mledger/LedgerOffloader.java

@@ -218,5 +218,17 @@ default CompletableFuture<Void> deleteOffloaded(UUID uid,
      * Close the resources if necessary
      */
     void close();
+
+    /**
+     * Scans all the ManagedLedgers stored on this Offloader (usually a Bucket).
+     * The callback should not modify/delete the ledgers.
+     * @param consumer receives the
+     * @param offloadDriverMetadata additional metadata
+     * @throws ManagedLedgerException
+     */
+    default void scanLedgers(OffloadedLedgerMetadataConsumer consumer,
+                             Map<String, String> offloadDriverMetadata) throws ManagedLedgerException {
+        throw ManagedLedgerException.getManagedLedgerException(new UnsupportedOperationException());
+    }
 }
 

managed-ledger/src/main/java/org/apache/bookkeeper/mledger/ManagedLedgerInfo.java

@@ -45,6 +45,7 @@ public static class LedgerInfo {
         public Long size;
         public Long timestamp;
         public boolean isOffloaded;
+        public String offloadedContextUuid;
     }
 
     public static class CursorInfo {

managed-ledger/src/main/java/org/apache/bookkeeper/mledger/OffloadedLedgerMetadata.java

@@ -0,0 +1,39 @@
+/**
+ * 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.bookkeeper.mledger;
+
+import lombok.Builder;
+import lombok.Getter;
+import lombok.ToString;
+
+import java.util.Map;
+
+@Builder
+@Getter
+@ToString
+public final class OffloadedLedgerMetadata {
+    private final long ledgerId;
+    private final long lastModified;
+    private final String name;
+    private final String bucketName;
+    private final String uri;
+    private final long size;
+    private final String uuid;
+    private final Map<String, String> userMetadata;
+}

managed-ledger/src/main/java/org/apache/bookkeeper/mledger/OffloadedLedgerMetadataConsumer.java

@@ -0,0 +1,32 @@
+/**
+ * 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.bookkeeper.mledger;
+
+public interface OffloadedLedgerMetadataConsumer {
+
+    /**
+     * Accepts metadata about a ledger.
+     * This function should not modify the ledger or delete it.
+     *
+     * @param ledger metadata
+     * @throws Exception
+     * @return return false in order to gracefully stop the scan
+     */
+    boolean accept(OffloadedLedgerMetadata ledger) throws Exception;
+}

managed-ledger/src/main/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerFactoryImpl.java

@@ -28,6 +28,7 @@
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
+import java.util.UUID;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CompletionException;
 import java.util.concurrent.ConcurrentHashMap;
@@ -593,6 +594,11 @@ public void operationComplete(MLDataFormats.ManagedLedgerInfo pbInfo, Stat stat)
                     ledgerInfo.entries = pbLedgerInfo.hasEntries() ? pbLedgerInfo.getEntries() : null;
                     ledgerInfo.size = pbLedgerInfo.hasSize() ? pbLedgerInfo.getSize() : null;
                     ledgerInfo.isOffloaded = pbLedgerInfo.hasOffloadContext();
+                    if (pbLedgerInfo.hasOffloadContext()) {
+                        MLDataFormats.OffloadContext offloadContext = pbLedgerInfo.getOffloadContext();
+                        UUID uuid = new UUID(offloadContext.getUidMsb(), offloadContext.getUidLsb());
+                        ledgerInfo.offloadedContextUuid = uuid.toString();
+                    }
                     info.ledgers.add(ledgerInfo);
                 }
 

pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespacesBase.java

@@ -31,7 +31,9 @@
 import java.lang.reflect.Field;
 import java.net.URI;
 import java.net.URL;
+import java.util.ArrayList;
 import java.util.Collections;
+import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Map;
@@ -40,10 +42,12 @@
 import java.util.Optional;
 import java.util.Set;
 import java.util.SortedSet;
+import java.util.TreeMap;
 import java.util.TreeSet;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Function;
 import java.util.stream.Collectors;
@@ -52,6 +56,10 @@
 import javax.ws.rs.core.Response;
 import javax.ws.rs.core.Response.Status;
 import javax.ws.rs.core.UriBuilder;
+import org.apache.bookkeeper.mledger.LedgerOffloader;
+import org.apache.bookkeeper.mledger.ManagedLedgerException;
+import org.apache.bookkeeper.mledger.ManagedLedgerFactory;
+import org.apache.bookkeeper.mledger.ManagedLedgerInfo;
 import org.apache.commons.lang.mutable.MutableObject;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.pulsar.broker.PulsarServerException;
@@ -2766,6 +2774,111 @@ protected void internalSetNamespaceResourceGroup(String rgName) {
         internalSetPolicies("resource_group_name", rgName);
     }
 
+    protected Map<String, Object> internalScanOffloadedLedgers() throws Exception {
+        log.info("internalScanOffloadedLedgers {}", namespaceName);
+        validateNamespacePolicyOperation(namespaceName, PolicyName.OFFLOAD, PolicyOperation.READ);
+
+        Policies policies = getNamespacePolicies(namespaceName);
+        LedgerOffloader managedLedgerOffloader = pulsar()
+                .getManagedLedgerOffloader(namespaceName, (OffloadPoliciesImpl) policies.offload_policies);
+
+        String localClusterName = pulsar().getConfiguration().getClusterName();
+        Map<String, Object> topLevelResult = new HashMap<>();
+        List<Map<String, Object>> objects = new ArrayList<>();
+        topLevelResult.put("objects", objects);
+        AtomicInteger totalCount = new AtomicInteger();
+        AtomicInteger totalErrors = new AtomicInteger();
+        AtomicInteger totalUnknown = new AtomicInteger();
+        managedLedgerOffloader.scanLedgers((md -> {
+            log.info("Found ledger {}", md);
+            Map<String, Object> objectInfo = new HashMap<>();
+            objectInfo.put("ledger", md.getLedgerId());
+            objectInfo.put("name", md.getName());
+            objectInfo.put("uri", md.getUri());
+            objectInfo.put("uuid", md.getUuid());
+            objectInfo.put("size", md.getSize());
+            objectInfo.put("lastModified", md.getLastModified());
+            objectInfo.put("userMetadata", md.getUserMetadata());
+
+            String status = "UNKNOWN";
+
+            if (md.getUserMetadata() != null) {
+                // non case sensistive
+                TreeMap<String, String> userMetadata = new TreeMap<>(String.CASE_INSENSITIVE_ORDER);
+                userMetadata.putAll(md.getUserMetadata());
+                String clusterName = userMetadata.get(LedgerOffloader.METADATA_PULSAR_CLUSTER_NAME);
+                if (localClusterName.equals(clusterName)) {
+                    String managedLedgerName = userMetadata.get("managedledgername");
+                    if (managedLedgerName != null) {
+                        objectInfo.put("managedLedgerName", managedLedgerName);
+                        try {
+                            status = checkLedgerShouldBeOnTieredStorage(md.getLedgerId(), md.getUuid(),
+                                    managedLedgerName, objectInfo, pulsar().getManagedLedgerFactory());
+                        } catch (InterruptedException err) {
+                            Thread.currentThread().interrupt();
+                            throw new RuntimeException(err);
+                        } catch (ManagedLedgerException err) {
+                            log.error("Error while checking managed ledger {}", managedLedgerName);
+                            throw new RuntimeException(err);
+                        }
+                    }
+                }
+            }
+            totalCount.incrementAndGet();
+            objectInfo.put("status", status);
+            switch (status) {
+                case "OK":
+                    break;
+                case "UNKNOWN":
+                    totalUnknown.incrementAndGet();
+                    break;
+                default:
+                    totalErrors.incrementAndGet();
+                    break;
+            }
+
+            objects.add(objectInfo);
+            return true;
+        }), managedLedgerOffloader.getOffloadDriverMetadata());
+        topLevelResult.put("errors", totalErrors.intValue());
+        topLevelResult.put("total", totalCount.intValue());
+        topLevelResult.put("unknownObjects", totalUnknown.intValue());
+        log.info("internalScanOffloadedLedgers {} scan finished");
+
+        return topLevelResult;
+    }
+
+    private static String checkLedgerShouldBeOnTieredStorage(long ledgerId, String uuid,
+                                                      String managedLedgerName,
+                                                      Map<String, Object> data,
+                                                      ManagedLedgerFactory managedLedgerFactory)
+            throws InterruptedException, ManagedLedgerException  {
+        try {
+            ManagedLedgerInfo managedLedgerInfo = managedLedgerFactory.getManagedLedgerInfo(managedLedgerName);
+            ManagedLedgerInfo.LedgerInfo ledgerInfo = managedLedgerInfo
+                    .ledgers.stream().filter(l -> l.ledgerId == ledgerId).findAny().orElse(null);
+            if (ledgerInfo == null) {
+                log.info("Managed ledger {} does not contain ledger {}", managedLedgerName, ledgerId);
+                return "NOT-FOUND";
+            }
+            data.put("numEntries", ledgerInfo.entries);
+            data.put("offloaded", ledgerInfo.isOffloaded);
+            if (!ledgerInfo.isOffloaded) {
+                log.info("Ledger {} is not marked as OFFLOADED in {}", ledgerId, managedLedgerName);
+                return "NOT-OFFLOADED";
+            }
+            String uuidOnMetadata = ledgerInfo.offloadedContextUuid;
+            if (!Objects.equals(uuidOnMetadata, uuid)) {
+                log.info("Ledger {} uuid {} does not match name uuid {}", ledgerId, uuidOnMetadata, uuid);
+                return "BAD-UUID";
+            }
+            return "OK";
+        }  catch (ManagedLedgerException.ManagedLedgerNotFoundException
+                | ManagedLedgerException.MetadataNotFoundException notFound) {
+            log.info("Managed ledger {} does not exist (maybe the topic has been deleted)", managedLedgerName);
+            return "NOT-FOUND";
+        }
+    }
 
     private static final Logger log = LoggerFactory.getLogger(NamespacesBase.class);
 }

pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/Namespaces.java

@@ -41,6 +41,7 @@
 import javax.ws.rs.container.AsyncResponse;
 import javax.ws.rs.container.Suspended;
 import javax.ws.rs.core.MediaType;
+import javax.ws.rs.core.Response;
 import org.apache.pulsar.broker.admin.impl.NamespacesBase;
 import org.apache.pulsar.broker.web.RestException;
 import org.apache.pulsar.client.api.SubscriptionType;
@@ -1871,5 +1872,22 @@ public void removeNamespaceResourceGroup(@PathParam("tenant") String tenant,
         internalSetNamespaceResourceGroup(null);
     }
 
+    @GET
+    @Path("/{tenant}/{namespace}/scanOffloadedLedgers")
+    @ApiOperation(value = "Trigger the scan of offloaded Ledgers on the LedgerOffloader for the given namespace")
+    @ApiResponses(value = { @ApiResponse(code = 403, message = "Don't have admin permission"),
+            @ApiResponse(code = 404, message = "Namespace doesn't exist") })
+    public Map<String, Object> scanOffloadedLedgers(@PathParam("tenant") String tenant,
+            @PathParam("namespace") String namespace) {
+        validateNamespaceName(tenant, namespace);
+        try {
+            return internalScanOffloadedLedgers();
+        } catch (Throwable err) {
+            log.error("Error while scanning offloaded ledgers for namespace {}", namespaceName, err);
+            throw new RestException(Response.Status.INTERNAL_SERVER_ERROR,
+                    "Error while scanning ledgers for " + namespaceName);
+        }
+    }
+
     private static final Logger log = LoggerFactory.getLogger(Namespaces.class);
 }