Skip to content

Commit e023ee7

Browse files
jjezrajjezra
committed
Add API get/clear heartbeats
1 parent f318f65 commit e023ee7

File tree

3 files changed

+103
-5
lines changed

3 files changed

+103
-5
lines changed

‎fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/IndexingBase.java

Lines changed: 10 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -1102,6 +1102,16 @@ boolean performIndexingStampOperation(@Nonnull ConcurrentHashMap<String, IndexBu
11021102
return true;
11031103
}
11041104

1105+
public CompletableFuture<Map<UUID, IndexBuildProto.IndexingHeartbeat>> getIndexingHeartbeats(int maxCount) {
1106+
return getRunner().runAsync(context -> openRecordStore(context)
1107+
.thenCompose(store -> IndexingHeartbeat.getIndexingHeartbeats(store, common.getPrimaryIndex(), maxCount)));
1108+
}
1109+
1110+
public CompletableFuture<Integer> clearIndexingHeartbeats(long minAgenMilliseconds, int maxIteration) {
1111+
return getRunner().runAsync(context -> openRecordStore(context)
1112+
.thenCompose(store -> IndexingHeartbeat.clearIndexingHeartbeats(store, common.getPrimaryIndex(), minAgenMilliseconds, maxIteration)));
1113+
}
1114+
11051115
/**
11061116
* Thrown when the indexing process fails to meet a precondition.
11071117
*/

‎fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/IndexingHeartbeat.java

Lines changed: 69 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -31,8 +31,11 @@
3131
import com.google.protobuf.InvalidProtocolBufferException;
3232

3333
import javax.annotation.Nonnull;
34+
import java.util.HashMap;
35+
import java.util.Map;
3436
import java.util.UUID;
3537
import java.util.concurrent.CompletableFuture;
38+
import java.util.concurrent.atomic.AtomicInteger;
3639

3740
public class IndexingHeartbeat {
3841
// [prefix, xid] -> [indexing-type, genesis time, heartbeat time]
@@ -74,7 +77,7 @@ public CompletableFuture<Void> checkAndUpdateHeartbeat(@Nonnull FDBRecordStore s
7477
if (!hasNext) {
7578
return false;
7679
}
77-
validateNonCompetingHeartbeat(iterator.next());
80+
validateNonCompetingHeartbeat(iterator.next(), nowMilliseconds());
7881
return true;
7982
}));
8083

@@ -84,13 +87,12 @@ public CompletableFuture<Void> checkAndUpdateHeartbeat(@Nonnull FDBRecordStore s
8487
}
8588
}
8689

87-
private void validateNonCompetingHeartbeat(KeyValue kv) {
90+
private void validateNonCompetingHeartbeat(KeyValue kv, long now) {
8891
final Tuple keyTuple = Tuple.fromBytes(kv.getKey());
8992
if (keyTuple.size() < 2) { // expecting 8
9093
return;
9194
}
9295
final UUID otherSessionId = keyTuple.getUUID(keyTuple.size() - 1);
93-
final long now = nowMilliseconds();
9496
if (!otherSessionId.equals(this.sessionId)) {
9597
try {
9698
final IndexBuildProto.IndexingHeartbeat otherHeartbeat = IndexBuildProto.IndexingHeartbeat.parseFrom(kv.getValue());
@@ -101,7 +103,6 @@ private void validateNonCompetingHeartbeat(KeyValue kv) {
101103
.addLogInfo(LogMessageKeys.EXISTING_SESSION_ID, otherSessionId)
102104
.addLogInfo(LogMessageKeys.AGE_MILLISECONDS, age)
103105
.addLogInfo(LogMessageKeys.TIME_LIMIT_MILLIS, leaseLength);
104-
// TODO: log details
105106
}
106107
} catch (InvalidProtocolBufferException e) {
107108
throw new RuntimeException(e);
@@ -113,8 +114,71 @@ public void clearHeartbeat(@Nonnull FDBRecordStore store, @Nonnull Index index)
113114
store.ensureContextActive().clear(IndexingSubspaces.indexheartbeatSubspace(store, index, sessionId).pack());
114115
}
115116

117+
public static CompletableFuture<Map<UUID, IndexBuildProto.IndexingHeartbeat>> getIndexingHeartbeats(FDBRecordStore store, Index index, int maxCount) {
118+
final Map<UUID, IndexBuildProto.IndexingHeartbeat> ret = new HashMap<>();
119+
final AsyncIterator<KeyValue> iterator = heartbeatsIterator(store, index);
120+
final AtomicInteger iterationCount = new AtomicInteger(0);
121+
return AsyncUtil.whileTrue(() -> iterator.onHasNext()
122+
.thenApply(hasNext -> {
123+
if (!hasNext) {
124+
return false;
125+
}
126+
if (maxCount > 0 && maxCount < iterationCount.incrementAndGet()) {
127+
return false;
128+
}
129+
final KeyValue kv = iterator.next();
130+
final Tuple keyTuple = Tuple.fromBytes(kv.getKey());
131+
if (keyTuple.size() < 2) { // expecting 8
132+
return true; // ignore, next
133+
}
134+
final UUID otherSessionId = keyTuple.getUUID(keyTuple.size() - 1);
135+
try {
136+
final IndexBuildProto.IndexingHeartbeat otherHeartbeat = IndexBuildProto.IndexingHeartbeat.parseFrom(kv.getValue());
137+
ret.put(otherSessionId, otherHeartbeat);
138+
} catch (InvalidProtocolBufferException e) {
139+
// put a NONE heartbeat to indicate an invalid item
140+
ret.put(otherSessionId, IndexBuildProto.IndexingHeartbeat.newBuilder()
141+
.setMethod(IndexBuildProto.IndexBuildIndexingStamp.Method.NONE)
142+
.build());
143+
}
144+
return true;
145+
}))
146+
.thenApply(ignore -> ret);
147+
}
148+
149+
public static CompletableFuture<Integer> clearIndexingHeartbeats(@Nonnull FDBRecordStore store, @Nonnull Index index, long minAgenMilliseconds, int maxIteration) {
150+
final AsyncIterator<KeyValue> iterator = heartbeatsIterator(store, index);
151+
final AtomicInteger deleteCount = new AtomicInteger(0);
152+
final AtomicInteger iterationCount = new AtomicInteger(0);
153+
final long now = nowMilliseconds();
154+
return AsyncUtil.whileTrue(() -> iterator.onHasNext()
155+
.thenApply(hasNext -> {
156+
if (!hasNext) {
157+
return false;
158+
}
159+
if (maxIteration > 0 && maxIteration < iterationCount.incrementAndGet()) {
160+
return false;
161+
}
162+
final KeyValue kv = iterator.next();
163+
boolean shouldRemove;
164+
try {
165+
final IndexBuildProto.IndexingHeartbeat otherHeartbeat = IndexBuildProto.IndexingHeartbeat.parseFrom(kv.getValue());
166+
// remove heartbeat if too old
167+
shouldRemove = now + minAgenMilliseconds <= otherHeartbeat.getHeartbeatTimeMilliseconds();
168+
} catch (InvalidProtocolBufferException e) {
169+
// remove heartbeat if invalid
170+
shouldRemove = true;
171+
}
172+
if (shouldRemove) {
173+
store.ensureContextActive().clear(kv.getKey());
174+
deleteCount.incrementAndGet();
175+
}
176+
return true;
177+
}))
178+
.thenApply(ignore -> deleteCount.get());
179+
}
116180

117-
public AsyncIterator<KeyValue> heartbeatsIterator(FDBRecordStore store, Index index) {
181+
public static AsyncIterator<KeyValue> heartbeatsIterator(FDBRecordStore store, Index index) {
118182
return store.getContext().ensureActive().snapshot().getRange(IndexingSubspaces.indexheartbeatSubspace(store, index).range()).iterator();
119183
}
120184

‎fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/OnlineIndexer.java

Lines changed: 24 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -50,6 +50,7 @@
5050
import java.util.List;
5151
import java.util.Map;
5252
import java.util.Set;
53+
import java.util.UUID;
5354
import java.util.concurrent.CompletableFuture;
5455
import java.util.concurrent.atomic.AtomicLong;
5556
import java.util.function.BiFunction;
@@ -596,6 +597,29 @@ private Map<String, IndexBuildProto.IndexBuildIndexingStamp> indexingStamp(@Null
596597
getIndexer().performIndexingStampOperation(op, id, ttlSeconds));
597598
}
598599

600+
/**
601+
* Get the current indexing heartbeats for a given index (single target or primary index).
602+
* @param maxCount safety valve to limit number items to read. Typically set to zero to keep unlimited.
603+
* @return map of session ids to {@link IndexBuildProto.IndexingHeartbeat}
604+
*/
605+
@API(API.Status.EXPERIMENTAL)
606+
public Map<UUID, IndexBuildProto.IndexingHeartbeat> getIndexingHeartbeats(int maxCount) {
607+
return asyncToSync(FDBStoreTimer.Waits.WAIT_INDEX_READ_HEARTBEATS,
608+
getIndexer().getIndexingHeartbeats(maxCount));
609+
}
610+
611+
/**
612+
* Clear old indexing heartbeats for a given index (single target or primary index).
613+
* @param minAgenMilliseconds minimum heartbeat age (in milliseconds) to clear.
614+
* @param maxIteration safety valve to limit number of items to check. Typically set to zero to keep unlimited
615+
* @return number of cleared heartbeats
616+
*/
617+
@API(API.Status.EXPERIMENTAL)
618+
public int clearIndexingHeartbeats(long minAgenMilliseconds, int maxIteration) {
619+
return asyncToSync(FDBStoreTimer.Waits.WAIT_INDEX_CLEAR_HEARTBEATS,
620+
getIndexer().clearIndexingHeartbeats(minAgenMilliseconds, maxIteration));
621+
}
622+
599623
/**
600624
* Wait for an asynchronous task to complete. This returns the result from the future or propagates
601625
* the error if the future completes exceptionally.

0 commit comments

Comments
 (0)