Skip to content

Commit

Permalink
decent amount of logging/clean up on thoth dynamic membership
Browse files Browse the repository at this point in the history 
For the moment, just disable the rebalancing test until I can pay some much needed attention to it.
  • Loading branch information
@Hellblazer
Hellblazer committed Dec 7, 2023
1 parent 7ede248 commit 5996e03
Show file tree
Hide file tree
Showing 6 changed files with 202 additions and 99 deletions.
9 changes: 9 additions & 0 deletions thoth/src/main/java/com/salesforce/apollo/thoth/KerlDHT.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -926,13 +926,19 @@ private void reconcile(Optional<Update> result,
log.debug("null interval reconciliation with {} : {} on: {}", destination.member().getId(),
member.getId(), e.getCause());
}
} else {
log.trace("Received no events in interval reconciliation from: {} on: {}", destination.member().getId(),
member.getId());
}
if (started.get()) {
scheduler.schedule(() -> reconcile(scheduler, duration), duration.toMillis(), TimeUnit.MILLISECONDS);
}
}

private Update reconcile(ReconciliationService link, Integer ring) {
if (member.equals(link.getMember())) {
return null;
}
CombinedIntervals keyIntervals = keyIntervals();
log.trace("Interval reconciliation on ring: {} with: {} on: {} intervals: {}", ring, link.getMember(),
member.getId(), keyIntervals);
Expand Down Expand Up @@ -1027,13 +1033,16 @@ private class Reconcile implements Reconciliation {
public Update reconcile(Intervals intervals, Digest from) {
var ring = intervals.getRing();
if (!valid(from, ring)) {
log.trace("Invalid reconcile from: {} ring: {} on: {}", from, ring, member.getId());
return Update.getDefaultInstance();
}
log.trace("Reconcile from: {} ring: {} on: {}", from, ring, member.getId());
try (var k = kerlPool.create()) {
final var builder = KerlDHT.this.kerlSpace.reconcile(intervals, k);
CombinedIntervals keyIntervals = keyIntervals();
builder.addAllIntervals(keyIntervals.toIntervals())
.setHave(kerlSpace.populate(Entropy.nextBitsStreamLong(), keyIntervals, fpr));
log.trace("Reconcile for: {} ring: {} count: {} on: {}", from, ring, builder.getEventsCount(), member);
return builder.build();
} catch (IOException | SQLException e) {
throw new IllegalStateException("Cannot acquire KERL", e);
Expand Down
14 changes: 9 additions & 5 deletions thoth/src/main/java/com/salesforce/apollo/thoth/KerlSpace.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -177,7 +177,7 @@ public static void upsert(DSLContext dsl, Validations validations) {
ULong.valueOf(coordinates.getSequenceNumber()).toBigInteger())
.returningResult(PENDING_COORDINATES.ID)
.fetchOne();
log.trace("Id: {} for: {}", id, logCoords);
log.trace("Id: {} for: {}", id.value1(), logCoords);
} catch (DataAccessException e) {
log.trace("access exception for: {}", logCoords, e);
// Already exists
Expand Down Expand Up @@ -215,6 +215,7 @@ public Biff populate(long seed, CombinedIntervals intervals, double fpr) {
try (var connection = connectionPool.getConnection()) {
var dsl = DSL.using(connection);
eventDigestsIn(intervals, dsl).forEach(d -> {
log.trace("Adding reconcile digest: {}", d);
bff.add(d);
});
} catch (SQLException e) {
Expand All @@ -238,14 +239,14 @@ public Update.Builder reconcile(Intervals intervals, DigestKERL kerl) {
var dsl = DSL.using(connection);
intervals.getIntervalsList()
.stream()
.map(i -> new KeyInterval(i))
.map(KeyInterval::new)
.flatMap(i -> eventDigestsIn(i, dsl))
.peek(d -> log.trace("reconcile digest: {}", d))
.filter(d -> !biff.contains(d))
.peek(d -> log.trace("filtered reconcile digest: {}", d))
.map(d -> event(d, dsl, kerl))
.filter(ke -> ke != null)
.forEach(ke -> {
update.addEvents(ke);
});
.forEach(update::addEvents);
} catch (SQLException e) {
log.error("Unable to provide estimated cardinality, cannot acquire JDBC connection", e);
throw new IllegalStateException("Unable to provide estimated cardinality, cannot acquire JDBC connection",
Expand All @@ -262,9 +263,11 @@ public Update.Builder reconcile(Intervals intervals, DigestKERL kerl) {
*/
public void update(List<KeyEventWithAttachmentAndValidations_> events, KERL kerl) {
if (events.isEmpty()) {
log.trace("No events to update");
return;
}

log.trace("Events to update: {}", events.size());
final var digestAlgorithm = kerl.getDigestAlgorithm();

try (var connection = connectionPool.getConnection()) {
Expand Down Expand Up @@ -301,6 +304,7 @@ private int cardinality() {
}

private void commitPending(DSLContext context, KERL kerl) {
log.trace("Commit pending");
context.select(PENDING_COORDINATES.ID, PENDING_EVENT.EVENT, PENDING_COORDINATES.ILK)
.from(PENDING_EVENT)
.join(PENDING_COORDINATES)
Expand Down
4 changes: 2 additions & 2 deletions thoth/src/test/java/com/salesforce/apollo/thoth/AbstractDhtTest.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -57,7 +57,7 @@ public class AbstractDhtTest {
protected static final boolean LARGE_TESTS = Boolean.getBoolean(
"large_tests");
protected static final double PBYZ = 0.25;
protected final Map<SigningMember, KerlDHT> dhts = new HashMap<>();
protected final Map<SigningMember, KerlDHT> dhts = new TreeMap<>();
protected final Map<SigningMember, Router> routers = new HashMap<>();
protected final AtomicBoolean gate = new AtomicBoolean(
false);
Expand Down Expand Up @@ -137,7 +137,7 @@ protected int getCardinality() {
protected void instantiate(SigningMember member, Context<Member> context,
ConcurrentSkipListMap<Digest, Member> serverMembers) {
context.activate(member);
final var url = String.format("jdbc:h2:mem:%s-%s;DB_CLOSE_DELAY=-1", member.getId(), prefix);
final var url = String.format("jdbc:h2:mem:%s-%s;DB_CLOSE_ON_EXIT=FALSE", member.getId(), prefix);
context.activate(member);
JdbcConnectionPool connectionPool = JdbcConnectionPool.create(url, "", "");
connectionPool.setMaxConnections(10);
Expand Down
137 changes: 106 additions & 31 deletions thoth/src/test/java/com/salesforce/apollo/thoth/DhtRebalanceTest.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -6,68 +6,143 @@
*/
package com.salesforce.apollo.thoth;

import com.salesforce.apollo.archipelago.LocalServer;
import com.salesforce.apollo.archipelago.Router;
import com.salesforce.apollo.archipelago.ServerConnectionCache;
import com.salesforce.apollo.cryptography.DigestAlgorithm;
import com.salesforce.apollo.stereotomy.EventCoordinates;
import com.salesforce.apollo.stereotomy.KERL;
import com.salesforce.apollo.stereotomy.Stereotomy;
import com.salesforce.apollo.stereotomy.StereotomyImpl;
import com.salesforce.apollo.membership.Context;
import com.salesforce.apollo.membership.Member;
import com.salesforce.apollo.membership.SigningMember;
import com.salesforce.apollo.membership.stereotomy.ControlledIdentifierMember;
import com.salesforce.apollo.stereotomy.*;
import com.salesforce.apollo.stereotomy.event.KeyEvent;
import com.salesforce.apollo.stereotomy.event.Seal.CoordinatesSeal;
import com.salesforce.apollo.stereotomy.event.Seal.DigestSeal;
import com.salesforce.apollo.stereotomy.identifier.SelfAddressingIdentifier;
import com.salesforce.apollo.stereotomy.identifier.spec.InteractionSpecification;
import com.salesforce.apollo.stereotomy.identifier.spec.RotationSpecification;
import com.salesforce.apollo.stereotomy.mem.MemKERL;
import com.salesforce.apollo.stereotomy.mem.MemKeyStore;
import com.salesforce.apollo.utils.Utils;
import org.h2.jdbcx.JdbcConnectionPool;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;

import java.security.SecureRandom;
import java.time.Duration;
import java.util.List;
import java.util.*;
import java.util.concurrent.Executors;
import java.util.stream.Collectors;
import java.util.stream.IntStream;

import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertTrue;

/**
* @author hal.hildebrand
*/
public class DhtRebalanceTest extends AbstractDhtTest {
private SecureRandom secureRandom;
public class DhtRebalanceTest {
public static final int CARDINALITY = 23;
private final TreeMap<SigningMember, Router> routers = new TreeMap<>();
private final TreeMap<SigningMember, KerlDHT> dhts = new TreeMap<>();
private final TreeMap<SigningMember, Context<Member>> contexts = new TreeMap<>();
private String prefix;
private SecureRandom entropy;
private StereotomyImpl stereotomy;
private MemKERL kerl;
private Map<ControlledIdentifierMember, ControlledIdentifier<SelfAddressingIdentifier>> identities;

@AfterEach
public void afterIt() throws Exception {
routers.values().forEach(r -> r.close(Duration.ofSeconds(1)));
routers.clear();
dhts.clear();
contexts.clear();
if (identities != null) {
identities.clear();
}
}

@BeforeEach
public void beforeIt() throws Exception {
secureRandom = SecureRandom.getInstance("SHA1PRNG");
secureRandom.setSeed(new byte[] { 0 });
entropy = SecureRandom.getInstance("SHA1PRNG");
entropy.setSeed(new byte[] { 6, 6, 6 });
prefix = UUID.randomUUID().toString();
kerl = new MemKERL(DigestAlgorithm.DEFAULT);
stereotomy = new StereotomyImpl(new MemKeyStore(), kerl, entropy);
identities = IntStream.range(0, CARDINALITY)
.mapToObj(i -> stereotomy.newIdentifier())
.collect(Collectors.toMap(controlled -> new ControlledIdentifierMember(controlled),
controlled -> controlled));
identities.keySet().forEach(member -> instantiate(member));
}

@Test
// @Test
public void lifecycle() throws Exception {
routers.values().forEach(r -> r.start());
dhts.values().forEach(dht -> dht.start(Duration.ofSeconds(1)));
var members = new TreeSet<SigningMember>();
var order = dhts.navigableKeySet().stream().toList();
System.out.println("Order: " + order);
members.add(order.getFirst());
KERL fristKerl = dhts.get(order.getFirst()).asKERL();
dhts.get(order.getFirst()).start(Duration.ofMillis(10));

Stereotomy controller = new StereotomyImpl(new MemKeyStore(), fristKerl, entropy);

KERL kerl = dhts.values().stream().findFirst().get().asKERL();
var identifier = controller.newIdentifier();
List<KERL.EventWithAttachments> identifierKerl = fristKerl.kerl(identifier.getIdentifier());
assertEquals(1, identifierKerl.size());
assertEquals(KeyEvent.INCEPTION_TYPE, identifierKerl.get(0).event().getIlk());

Stereotomy controller = new StereotomyImpl(new MemKeyStore(), kerl, secureRandom);
var remaining = order.subList(1, order.size());
members.add(remaining.getFirst());
var test = dhts.get(remaining.getFirst());
test.start(Duration.ofMillis(10));
var testKerl = test.asKERL();
members.forEach(m -> {
contexts.values().forEach(c -> c.activate(m));
});

var i = controller.newIdentifier();
assertTrue(Utils.waitForCondition(20_000, 1000, () -> testKerl.kerl(identifier.getIdentifier()).size() == 1));
var mKerl = testKerl.kerl(identifier.getIdentifier());
assertEquals(1, mKerl.size());
assertEquals(KeyEvent.INCEPTION_TYPE, mKerl.get(0).event().getIlk());

var digest = DigestAlgorithm.BLAKE3_256.digest("digest seal".getBytes());
var event = EventCoordinates.of(kerl.getKeyEvent(i.getLastEstablishmentEvent()));
var event = EventCoordinates.of(testKerl.getKeyEvent(identifier.getLastEstablishmentEvent()));
var seals = List.of(DigestSeal.construct(digest), DigestSeal.construct(digest),
CoordinatesSeal.construct(event));

i.rotate();
i.seal(InteractionSpecification.newBuilder());
i.rotate(RotationSpecification.newBuilder().addAllSeals(seals));
i.seal(InteractionSpecification.newBuilder().addAllSeals(seals));
i.rotate();
i.rotate();
var iKerl = kerl.kerl(i.getIdentifier());
assertEquals(7, iKerl.size());
assertEquals(KeyEvent.INCEPTION_TYPE, iKerl.get(0).event().getIlk());
assertEquals(KeyEvent.ROTATION_TYPE, iKerl.get(1).event().getIlk());
assertEquals(KeyEvent.INTERACTION_TYPE, iKerl.get(2).event().getIlk());
assertEquals(KeyEvent.ROTATION_TYPE, iKerl.get(3).event().getIlk());
assertEquals(KeyEvent.INTERACTION_TYPE, iKerl.get(4).event().getIlk());
assertEquals(KeyEvent.ROTATION_TYPE, iKerl.get(5).event().getIlk());
assertEquals(KeyEvent.ROTATION_TYPE, iKerl.get(6).event().getIlk());
identifier.rotate();
identifier.seal(InteractionSpecification.newBuilder());
identifier.rotate(RotationSpecification.newBuilder().addAllSeals(seals));
identifier.seal(InteractionSpecification.newBuilder().addAllSeals(seals));
identifier.rotate();
identifier.rotate();

identifierKerl = testKerl.kerl(identifier.getIdentifier());
assertEquals(7, identifierKerl.size());
assertEquals(KeyEvent.INCEPTION_TYPE, identifierKerl.get(0).event().getIlk());
assertEquals(KeyEvent.ROTATION_TYPE, identifierKerl.get(1).event().getIlk());
assertEquals(KeyEvent.INTERACTION_TYPE, identifierKerl.get(2).event().getIlk());
assertEquals(KeyEvent.ROTATION_TYPE, identifierKerl.get(3).event().getIlk());
assertEquals(KeyEvent.INTERACTION_TYPE, identifierKerl.get(4).event().getIlk());
assertEquals(KeyEvent.ROTATION_TYPE, identifierKerl.get(5).event().getIlk());
assertEquals(KeyEvent.ROTATION_TYPE, identifierKerl.get(6).event().getIlk());
}

protected void instantiate(SigningMember member) {
var context = Context.<Member>newBuilder().build();
contexts.put(member, context);
context.activate(member);
final var url = String.format("jdbc:h2:mem:%s-%s;DB_CLOSE_ON_EXIT=FALSE", member.getId(), prefix);
context.activate(member);
JdbcConnectionPool connectionPool = JdbcConnectionPool.create(url, "", "");
connectionPool.setMaxConnections(10);
var exec = Executors.newVirtualThreadPerTaskExecutor();
var router = new LocalServer(prefix, member).router(ServerConnectionCache.newBuilder().setTarget(2));
routers.put(member, router);
dhts.put(member, new KerlDHT(Duration.ofMillis(3), context, member, (t, k) -> k, connectionPool,
DigestAlgorithm.DEFAULT, router, Duration.ofSeconds(1), 0.0125, null));
}
}
Loading

0 comments on commit 5996e03

Please sign in to comment.