Skip to content

Commit 6710a7b

Browse files
merlimatmerlimat
authored
Added REST and CLI tool to expose ManagedLedger metadata (#292)
1 parent 5a28f63 commit 6710a7b

File tree

11 files changed

+471
-4
lines changed

11 files changed

+471
-4
lines changed

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

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -110,4 +110,10 @@ public interface ResetCursorCallback {
110110
public void resetFailed(ManagedLedgerException exception, Object ctx);
111111
}
112112

113+
public interface ManagedLedgerInfoCallback {
114+
public void getInfoComplete(ManagedLedgerInfo info, Object ctx);
115+
116+
public void getInfoFailed(ManagedLedgerException exception, Object ctx);
117+
}
118+
113119
}

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

Lines changed: 25 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -15,6 +15,7 @@
1515
*/
1616
package org.apache.bookkeeper.mledger;
1717

18+
import org.apache.bookkeeper.mledger.AsyncCallbacks.ManagedLedgerInfoCallback;
1819
import org.apache.bookkeeper.mledger.AsyncCallbacks.OpenLedgerCallback;
1920

2021
import com.google.common.annotations.Beta;
@@ -78,6 +79,30 @@ public ManagedLedger open(String name, ManagedLedgerConfig config)
7879
*/
7980
public void asyncOpen(String name, ManagedLedgerConfig config, OpenLedgerCallback callback, Object ctx);
8081

82+
/**
83+
* Get the current metadata info for a managed ledger
84+
*
85+
* @param name
86+
* the unique name that identifies the managed ledger
87+
* @param callback
88+
* callback object
89+
* @param ctx
90+
* opaque context
91+
*/
92+
public ManagedLedgerInfo getManagedLedgerInfo(String name) throws InterruptedException, ManagedLedgerException;
93+
94+
/**
95+
* Asynchronously get the current metadata info for a managed ledger
96+
*
97+
* @param name
98+
* the unique name that identifies the managed ledger
99+
* @param callback
100+
* callback object
101+
* @param ctx
102+
* opaque context
103+
*/
104+
public void asyncGetManagedLedgerInfo(String name, ManagedLedgerInfoCallback callback, Object ctx);
105+
81106
/**
82107
* Releases all the resources maintained by the ManagedLedgerFactory
83108
*

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

Lines changed: 75 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,75 @@
1+
/**
2+
* Copyright 2016 Yahoo Inc.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License");
5+
* you may not use this file except in compliance with the License.
6+
* You may obtain a copy of the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS,
12+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13+
* See the License for the specific language governing permissions and
14+
* limitations under the License.
15+
*/
16+
package org.apache.bookkeeper.mledger;
17+
18+
import java.util.List;
19+
import java.util.Map;
20+
21+
public class ManagedLedgerInfo {
22+
/** Z-Node version */
23+
public int version;
24+
public String creationDate;
25+
public String modificationDate;
26+
27+
public List<LedgerInfo> ledgers;
28+
29+
public Map<String, CursorInfo> cursors;
30+
31+
public static class LedgerInfo {
32+
public long ledgerId;
33+
public Long entries;
34+
public Long size;
35+
public Long timestamp;
36+
}
37+
38+
public static class CursorInfo {
39+
/** Z-Node version */
40+
public int version;
41+
public String creationDate;
42+
public String modificationDate;
43+
44+
// If the ledger id is -1, then the mark-delete position is
45+
// the one from the (ledgerId, entryId) snapshot below
46+
public long cursorsLedgerId;
47+
48+
// Last snapshot of the mark-delete position
49+
public PositionInfo markDelete;
50+
public List<MessageRangeInfo> individualDeletedMessages;
51+
}
52+
53+
public static class PositionInfo {
54+
public long ledgerId;
55+
public long entryId;
56+
57+
@Override
58+
public String toString() {
59+
return String.format("%d:%d", ledgerId, entryId);
60+
}
61+
}
62+
63+
public static class MessageRangeInfo {
64+
// Starting of the range (not included)
65+
public PositionInfo from = new PositionInfo();
66+
67+
// End of the range (included)
68+
public PositionInfo to = new PositionInfo();
69+
70+
@Override
71+
public String toString() {
72+
return String.format("(%s, %s]", from, to);
73+
}
74+
}
75+
}

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

Lines changed: 148 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -17,9 +17,15 @@
1717

1818
import static com.google.common.base.Preconditions.checkNotNull;
1919

20+
import java.time.Instant;
21+
import java.time.ZoneId;
22+
import java.time.format.DateTimeFormatter;
23+
import java.util.ArrayList;
24+
import java.util.List;
2025
import java.util.Map;
2126
import java.util.concurrent.CompletableFuture;
2227
import java.util.concurrent.ConcurrentHashMap;
28+
import java.util.concurrent.ConcurrentSkipListMap;
2329
import java.util.concurrent.CountDownLatch;
2430
import java.util.concurrent.Executors;
2531
import java.util.concurrent.ScheduledExecutorService;
@@ -30,15 +36,28 @@
3036
import org.apache.bookkeeper.client.BookKeeper;
3137
import org.apache.bookkeeper.conf.ClientConfiguration;
3238
import org.apache.bookkeeper.mledger.AsyncCallbacks;
39+
import org.apache.bookkeeper.mledger.AsyncCallbacks.ManagedLedgerInfoCallback;
3340
import org.apache.bookkeeper.mledger.AsyncCallbacks.OpenLedgerCallback;
3441
import org.apache.bookkeeper.mledger.ManagedLedger;
3542
import org.apache.bookkeeper.mledger.ManagedLedgerConfig;
3643
import org.apache.bookkeeper.mledger.ManagedLedgerException;
44+
import org.apache.bookkeeper.mledger.ManagedLedgerException.MetaStoreException;
3745
import org.apache.bookkeeper.mledger.ManagedLedgerFactory;
3846
import org.apache.bookkeeper.mledger.ManagedLedgerFactoryConfig;
3947
import org.apache.bookkeeper.mledger.ManagedLedgerFactoryMXBean;
48+
import org.apache.bookkeeper.mledger.ManagedLedgerInfo;
49+
import org.apache.bookkeeper.mledger.ManagedLedgerInfo.CursorInfo;
50+
import org.apache.bookkeeper.mledger.ManagedLedgerInfo.LedgerInfo;
51+
import org.apache.bookkeeper.mledger.ManagedLedgerInfo.MessageRangeInfo;
52+
import org.apache.bookkeeper.mledger.ManagedLedgerInfo.PositionInfo;
4053
import org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl.ManagedLedgerInitializeLedgerCallback;
4154
import org.apache.bookkeeper.mledger.impl.ManagedLedgerImpl.State;
55+
import org.apache.bookkeeper.mledger.impl.MetaStore.MetaStoreCallback;
56+
import org.apache.bookkeeper.mledger.impl.MetaStore.Stat;
57+
import org.apache.bookkeeper.mledger.proto.MLDataFormats;
58+
import org.apache.bookkeeper.mledger.proto.MLDataFormats.ManagedCursorInfo;
59+
import org.apache.bookkeeper.mledger.proto.MLDataFormats.MessageRange;
60+
import org.apache.bookkeeper.mledger.util.Futures;
4261
import org.apache.bookkeeper.util.OrderedSafeExecutor;
4362
import org.apache.zookeeper.Watcher;
4463
import org.apache.zookeeper.ZooKeeper;
@@ -290,6 +309,133 @@ public void closeFailed(ManagedLedgerException exception, Object ctx) {
290309
entryCacheManager.clear();
291310
}
292311

312+
@Override
313+
public ManagedLedgerInfo getManagedLedgerInfo(String name) throws InterruptedException, ManagedLedgerException {
314+
class Result {
315+
ManagedLedgerInfo info = null;
316+
ManagedLedgerException e = null;
317+
}
318+
final Result r = new Result();
319+
final CountDownLatch latch = new CountDownLatch(1);
320+
asyncGetManagedLedgerInfo(name, new ManagedLedgerInfoCallback() {
321+
@Override
322+
public void getInfoComplete(ManagedLedgerInfo info, Object ctx) {
323+
r.info = info;
324+
latch.countDown();
325+
}
326+
327+
@Override
328+
public void getInfoFailed(ManagedLedgerException exception, Object ctx) {
329+
r.e = exception;
330+
latch.countDown();
331+
}
332+
}, null);
333+
334+
latch.await();
335+
336+
if (r.e != null) {
337+
throw r.e;
338+
}
339+
return r.info;
340+
}
341+
342+
@Override
343+
public void asyncGetManagedLedgerInfo(String name, ManagedLedgerInfoCallback callback, Object ctx) {
344+
store.getManagedLedgerInfo(name, new MetaStoreCallback<MLDataFormats.ManagedLedgerInfo>() {
345+
@Override
346+
public void operationComplete(MLDataFormats.ManagedLedgerInfo pbInfo, Stat stat) {
347+
ManagedLedgerInfo info = new ManagedLedgerInfo();
348+
info.version = stat.getVersion();
349+
info.creationDate = DATE_FORMAT.format(Instant.ofEpochMilli(stat.getCreationTimestamp()));
350+
info.modificationDate = DATE_FORMAT.format(Instant.ofEpochMilli(stat.getModificationTimestamp()));
351+
352+
info.ledgers = new ArrayList<>(pbInfo.getLedgerInfoCount());
353+
for (int i = 0; i < pbInfo.getLedgerInfoCount(); i++) {
354+
MLDataFormats.ManagedLedgerInfo.LedgerInfo pbLedgerInfo = pbInfo.getLedgerInfo(i);
355+
LedgerInfo ledgerInfo = new LedgerInfo();
356+
ledgerInfo.ledgerId = pbLedgerInfo.getLedgerId();
357+
ledgerInfo.entries = pbLedgerInfo.hasEntries() ? pbLedgerInfo.getEntries() : null;
358+
ledgerInfo.size = pbLedgerInfo.hasSize() ? pbLedgerInfo.getSize() : null;
359+
info.ledgers.add(ledgerInfo);
360+
}
361+
362+
store.getCursors(name, new MetaStoreCallback<List<String>>() {
363+
@Override
364+
public void operationComplete(List<String> cursorsList, Stat stat) {
365+
// Get the info for each cursor
366+
info.cursors = new ConcurrentSkipListMap<>();
367+
List<CompletableFuture<Void>> cursorsFutures = new ArrayList<>();
368+
369+
for (String cursorName : cursorsList) {
370+
CompletableFuture<Void> cursorFuture = new CompletableFuture<>();
371+
cursorsFutures.add(cursorFuture);
372+
store.asyncGetCursorInfo(name, cursorName,
373+
new MetaStoreCallback<MLDataFormats.ManagedCursorInfo>() {
374+
@Override
375+
public void operationComplete(ManagedCursorInfo pbCursorInfo, Stat stat) {
376+
CursorInfo cursorInfo = new CursorInfo();
377+
cursorInfo.version = stat.getVersion();
378+
cursorInfo.creationDate = DATE_FORMAT
379+
.format(Instant.ofEpochMilli(stat.getCreationTimestamp()));
380+
cursorInfo.modificationDate = DATE_FORMAT
381+
.format(Instant.ofEpochMilli(stat.getModificationTimestamp()));
382+
383+
cursorInfo.cursorsLedgerId = pbCursorInfo.getCursorsLedgerId();
384+
385+
if (pbCursorInfo.hasMarkDeleteLedgerId()) {
386+
cursorInfo.markDelete = new PositionInfo();
387+
cursorInfo.markDelete.ledgerId = pbCursorInfo.getMarkDeleteLedgerId();
388+
cursorInfo.markDelete.entryId = pbCursorInfo.getMarkDeleteEntryId();
389+
}
390+
391+
if (pbCursorInfo.getIndividualDeletedMessagesCount() > 0) {
392+
cursorInfo.individualDeletedMessages = new ArrayList<>();
393+
for (int i = 0; i < pbCursorInfo
394+
.getIndividualDeletedMessagesCount(); i++) {
395+
MessageRange range = pbCursorInfo.getIndividualDeletedMessages(i);
396+
MessageRangeInfo rangeInfo = new MessageRangeInfo();
397+
rangeInfo.from.ledgerId = range.getLowerEndpoint().getLedgerId();
398+
rangeInfo.from.entryId = range.getLowerEndpoint().getEntryId();
399+
rangeInfo.to.ledgerId = range.getUpperEndpoint().getLedgerId();
400+
rangeInfo.to.entryId = range.getUpperEndpoint().getEntryId();
401+
cursorInfo.individualDeletedMessages.add(rangeInfo);
402+
}
403+
}
404+
405+
info.cursors.put(cursorName, cursorInfo);
406+
cursorFuture.complete(null);
407+
}
408+
409+
@Override
410+
public void operationFailed(MetaStoreException e) {
411+
cursorFuture.completeExceptionally(e);
412+
}
413+
});
414+
}
415+
416+
Futures.waitForAll(cursorsFutures).thenRun(() -> {
417+
// Completed all the cursors info
418+
callback.getInfoComplete(info, ctx);
419+
}).exceptionally((ex) -> {
420+
callback.getInfoFailed(new ManagedLedgerException(ex), ctx);
421+
return null;
422+
});
423+
}
424+
425+
@Override
426+
public void operationFailed(MetaStoreException e) {
427+
callback.getInfoFailed(e, ctx);
428+
}
429+
});
430+
}
431+
432+
@Override
433+
public void operationFailed(MetaStoreException e) {
434+
callback.getInfoFailed(e, ctx);
435+
}
436+
});
437+
}
438+
293439
public MetaStore getMetaStore() {
294440
return store;
295441
}
@@ -311,4 +457,6 @@ public BookKeeper getBookKeeper() {
311457
}
312458

313459
private static final Logger log = LoggerFactory.getLogger(ManagedLedgerFactoryImpl.class);
460+
461+
private static final DateTimeFormatter DATE_FORMAT = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss.SSSZ").withZone(ZoneId.systemDefault());
314462
}

0 commit comments

Comments
 (0)