Signal/Signal-Server
1
0
Fork 0
mirror of https://github.com/signalapp/Signal-Server synced 2026-04-20 07:28:06 +01:00
Code Issues Packages Projects Releases Wiki Activity

uuid-based account crawler

Browse Source
This commit is contained in:
Jeffrey Griffin
2019-08-26 14:11:23 -07:00
parent bb52049bf4
commit 69742839c0
17 changed files with 202 additions and 125 deletions
Download Patch File Download Diff File Expand all files Collapse all files

15
service/src/main/java/org/whispersystems/textsecuregcm/entities/ActiveUserTally.java
View File

@@ -19,10 +19,11 @@ package org.whispersystems.textsecuregcm.entities;
import com.fasterxml.jackson.annotation.JsonProperty;
import java.util.Map;
import java.util.HashMap;
import java.util.UUID;
public class ActiveUserTally {
@JsonProperty
private String fromNumber;
private UUID fromUuid;
@JsonProperty
private Map<String, long[]> platforms;
@@ -32,14 +33,14 @@ public class ActiveUserTally {
public ActiveUserTally() {}
public ActiveUserTally(String fromNumber, Map<String, long[]> platforms, Map<String, long[]> countries) {
this.fromNumber = fromNumber;
public ActiveUserTally(UUID fromUuid, Map<String, long[]> platforms, Map<String, long[]> countries) {
this.fromUuid = fromUuid;
this.platforms = platforms;
this.countries = countries;
}
public String getFromNumber() {
return this.fromNumber;
public UUID getFromUuid() {
return this.fromUuid;
}
public Map<String, long[]> getPlatforms() {
@@ -50,8 +51,8 @@ public class ActiveUserTally {
return this.countries;
}
public void setFromNumber(String fromNumber) {
this.fromNumber = fromNumber;
public void setFromUuid(UUID fromUuid) {
this.fromUuid = fromUuid;
}
}

72
service/src/main/java/org/whispersystems/textsecuregcm/entities/DirectoryReconciliationRequest.java
View File

@@ -18,38 +18,84 @@ package org.whispersystems.textsecuregcm.entities;
import com.fasterxml.jackson.annotation.JsonProperty;
import javax.validation.constraints.NotNull;
import java.util.List;
import java.util.UUID;
public class DirectoryReconciliationRequest {
@JsonProperty
private String fromNumber;
private UUID fromUuid;
@JsonProperty
private String toNumber;
private UUID toUuid;
@JsonProperty
private List<String> numbers;
private List<User> users;
public DirectoryReconciliationRequest() {
}
public DirectoryReconciliationRequest(String fromNumber, String toNumber, List<String> numbers) {
this.fromNumber = fromNumber;
this.toNumber = toNumber;
this.numbers = numbers;
public DirectoryReconciliationRequest(UUID fromUuid, UUID toUuid, List<User> users) {
this.fromUuid = fromUuid;
this.toUuid = toUuid;
this.users = users;
}
public String getFromNumber() {
return fromNumber;
public UUID getFromUuid() {
return fromUuid;
}
public String getToNumber() {
return toNumber;
public UUID getToUuid() {
return toUuid;
}
public List<String> getNumbers() {
return numbers;
public List<User> getUsers() {
return users;
}
public static class User {
@JsonProperty
private UUID uuid;
@JsonProperty
private String number;
public User() {
}
public User(UUID uuid, String number) {
this.uuid = uuid;
this.number = number;
}
public UUID getUuid() {
return uuid;
}
public String getNumber() {
return number;
}
@Override
public boolean equals(Object o) {
if (this == o) return true;
if (o == null || getClass() != o.getClass()) return false;
User user = (User) o;
if (uuid != null ? !uuid.equals(user.uuid) : user.uuid != null) return false;
if (number != null ? !number.equals(user.number) : user.number != null) return false;
return true;
}
@Override
public int hashCode() {
int result = uuid != null ? uuid.hashCode() : 0;
result = 31 * result + (number != null ? number.hashCode() : 0);
return result;
}
}
}

5
service/src/main/java/org/whispersystems/textsecuregcm/storage/AccountCleaner.java
View File

@@ -26,6 +26,7 @@ import org.whispersystems.textsecuregcm.util.Util;
import java.util.List;
import java.util.Optional;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
import static com.codahale.metrics.MetricRegistry.name;
@@ -51,7 +52,7 @@ public class AccountCleaner implements AccountDatabaseCrawlerListener {
}
@Override
public void onCrawlChunk(Optional<String> fromNumber, List<Account> chunkAccounts) {
public void onCrawlChunk(Optional<UUID> fromUuid, List<Account> chunkAccounts) {
int accountUpdateCount = 0;
for (Account account : chunkAccounts) {
if (needsExplicitRemoval(account)) {
@@ -74,7 +75,7 @@ public class AccountCleaner implements AccountDatabaseCrawlerListener {
}
@Override
public void onCrawlEnd(Optional<String> fromNumber) {
public void onCrawlEnd(Optional<UUID> fromUuid) {
}
private boolean needsExplicitRemoval(Account account) {

24
service/src/main/java/org/whispersystems/textsecuregcm/storage/AccountDatabaseCrawler.java
View File

@@ -122,26 +122,26 @@ public class AccountDatabaseCrawler implements Managed, Runnable {
}
private void processChunk() {
Optional<String> fromNumber = cache.getLastNumber();
Optional<UUID> fromUuid = cache.getLastUuid();
if (!fromNumber.isPresent()) {
listeners.forEach(listener -> { listener.onCrawlStart(); });
if (!fromUuid.isPresent()) {
listeners.forEach(AccountDatabaseCrawlerListener::onCrawlStart);
}
List<Account> chunkAccounts = readChunk(fromNumber, chunkSize);
List<Account> chunkAccounts = readChunk(fromUuid, chunkSize);
if (chunkAccounts.isEmpty()) {
listeners.forEach(listener -> { listener.onCrawlEnd(fromNumber); });
cache.setLastNumber(Optional.empty());
listeners.forEach(listener -> listener.onCrawlEnd(fromUuid));
cache.setLastUuid(Optional.empty());
cache.clearAccelerate();
} else {
try {
for (AccountDatabaseCrawlerListener listener : listeners) {
listener.onCrawlChunk(fromNumber, chunkAccounts);
listener.onCrawlChunk(fromUuid, chunkAccounts);
}
cache.setLastNumber(Optional.of(chunkAccounts.get(chunkAccounts.size() - 1).getNumber()));
cache.setLastUuid(Optional.of(chunkAccounts.get(chunkAccounts.size() - 1).getUuid()));
} catch (AccountDatabaseCrawlerRestartException e) {
cache.setLastNumber(Optional.empty());
cache.setLastUuid(Optional.empty());
cache.clearAccelerate();
}
@@ -149,12 +149,12 @@ public class AccountDatabaseCrawler implements Managed, Runnable {
}
private List<Account> readChunk(Optional<String> fromNumber, int chunkSize) {
private List<Account> readChunk(Optional<UUID> fromUuid, int chunkSize) {
try (Timer.Context timer = readChunkTimer.time()) {
List<Account> chunkAccounts;
if (fromNumber.isPresent()) {
chunkAccounts = accounts.getAllFrom(fromNumber.get(), chunkSize);
if (fromUuid.isPresent()) {
chunkAccounts = accounts.getAllFrom(fromUuid.get(), chunkSize);
} else {
chunkAccounts = accounts.getAllFrom(chunkSize);
}

18
service/src/main/java/org/whispersystems/textsecuregcm/storage/AccountDatabaseCrawlerCache.java
View File

@@ -23,6 +23,7 @@ import java.io.IOException;
import java.util.Arrays;
import java.util.List;
import java.util.Optional;
import java.util.UUID;
import redis.clients.jedis.Jedis;
@@ -30,7 +31,7 @@ import redis.clients.jedis.Jedis;
public class AccountDatabaseCrawlerCache {
private static final String ACTIVE_WORKER_KEY = "account_database_crawler_cache_active_worker";
private static final String LAST_NUMBER_KEY = "account_database_crawler_cache_last_number";
private static final String LAST_UUID_KEY = "account_database_crawler_cache_last_uuid";
private static final String ACCELERATE_KEY = "account_database_crawler_cache_accelerate";
private static final long LAST_NUMBER_TTL_MS = 86400_000L;
@@ -67,18 +68,21 @@ public class AccountDatabaseCrawlerCache {
luaScript.execute(keys, args);
}
public Optional<String> getLastNumber() {
public Optional<UUID> getLastUuid() {
try (Jedis jedis = jedisPool.getWriteResource()) {
return Optional.ofNullable(jedis.get(LAST_NUMBER_KEY));
String lastUuidString = jedis.get(LAST_UUID_KEY);
if (lastUuidString == null) return Optional.empty();
else return Optional.of(UUID.fromString(lastUuidString));
}
}
public void setLastNumber(Optional<String> lastNumber) {
public void setLastUuid(Optional<UUID> lastUuid) {
try (Jedis jedis = jedisPool.getWriteResource()) {
if (lastNumber.isPresent()) {
jedis.psetex(LAST_NUMBER_KEY, LAST_NUMBER_TTL_MS, lastNumber.get());
if (lastUuid.isPresent()) {
jedis.psetex(LAST_UUID_KEY, LAST_NUMBER_TTL_MS, lastUuid.get().toString());
} else {
jedis.del(LAST_NUMBER_KEY);
jedis.del(LAST_UUID_KEY);
}
}
}

5
service/src/main/java/org/whispersystems/textsecuregcm/storage/AccountDatabaseCrawlerListener.java
View File

@@ -18,10 +18,11 @@ package org.whispersystems.textsecuregcm.storage;
import java.util.List;
import java.util.Optional;
import java.util.UUID;
@SuppressWarnings("OptionalUsedAsFieldOrParameterType")
public interface AccountDatabaseCrawlerListener {
void onCrawlStart();
void onCrawlChunk(Optional<String> fromNumber, List<Account> chunkAccounts) throws AccountDatabaseCrawlerRestartException;
void onCrawlEnd(Optional<String> fromNumber);
void onCrawlChunk(Optional<UUID> fromUuid, List<Account> chunkAccounts) throws AccountDatabaseCrawlerRestartException;
void onCrawlEnd(Optional<UUID> fromUuid);
}

6
service/src/main/java/org/whispersystems/textsecuregcm/storage/Accounts.java
View File

@@ -111,10 +111,10 @@ public class Accounts {
}));
}
public List<Account> getAllFrom(String from, int length) {
public List<Account> getAllFrom(UUID from, int length) {
return database.with(jdbi -> jdbi.withHandle(handle -> {
try (Timer.Context ignored = getAllFromOffsetTimer.time()) {
return handle.createQuery("SELECT * FROM accounts WHERE " + NUMBER + " > :from ORDER BY " + NUMBER + " LIMIT :limit")
return handle.createQuery("SELECT * FROM accounts WHERE " + UID + " > :from ORDER BY " + UID + " LIMIT :limit")
.bind("from", from)
.bind("limit", length)
.mapTo(Account.class)
@@ -126,7 +126,7 @@ public class Accounts {
public List<Account> getAllFrom(int length) {
return database.with(jdbi -> jdbi.withHandle(handle -> {
try (Timer.Context ignored = getAllFromTimer.time()) {
return handle.createQuery("SELECT * FROM accounts ORDER BY " + NUMBER + " LIMIT :limit")
return handle.createQuery("SELECT * FROM accounts ORDER BY " + UID + " LIMIT :limit")
.bind("limit", length)
.mapTo(Account.class)
.list();

4
service/src/main/java/org/whispersystems/textsecuregcm/storage/AccountsManager.java
View File

@@ -121,8 +121,8 @@ public class AccountsManager {
return accounts.getAllFrom(length);
}
public List<Account> getAllFrom(String number, int length) {
return accounts.getAllFrom(number, length);
public List<Account> getAllFrom(UUID uuid, int length) {
return accounts.getAllFrom(uuid, length);
}
private void updateDirectory(Account account) {

19
service/src/main/java/org/whispersystems/textsecuregcm/storage/ActiveUserCounter.java
View File

@@ -30,6 +30,7 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
import io.dropwizard.metrics.MetricsFactory;
@@ -43,8 +44,6 @@ public class ActiveUserCounter implements AccountDatabaseCrawlerListener {
private static final String PLATFORM_IOS = "ios";
private static final String PLATFORM_ANDROID = "android";
private static final String FIRST_FROM_NUMBER = "+";
private static final String INTERVALS[] = {"daily", "weekly", "monthly", "quarterly", "yearly"};
private final MetricsFactory metricsFactory;
@@ -64,7 +63,7 @@ public class ActiveUserCounter implements AccountDatabaseCrawlerListener {
}
@Override
public void onCrawlChunk(Optional<String> fromNumber, List<Account> chunkAccounts) {
public void onCrawlChunk(Optional<UUID> fromNumber, List<Account> chunkAccounts) {
long nowDays = TimeUnit.MILLISECONDS.toDays(System.currentTimeMillis());
long agoMs[] = {TimeUnit.DAYS.toMillis(nowDays - 1),
TimeUnit.DAYS.toMillis(nowDays - 7),
@@ -107,12 +106,11 @@ public class ActiveUserCounter implements AccountDatabaseCrawlerListener {
}
}
incrementTallies(fromNumber.orElse(FIRST_FROM_NUMBER), platformIncrements, countryIncrements);
incrementTallies(fromNumber.orElse(UUID.randomUUID()), platformIncrements, countryIncrements);
}
@Override
public void onCrawlEnd(Optional<String> fromNumber) {
public void onCrawlEnd(Optional<UUID> fromNumber) {
MetricRegistry metrics = new MetricRegistry();
long intervalTallies[] = new long[INTERVALS.length];
ActiveUserTally activeUserTally = getFinalTallies();
@@ -156,17 +154,18 @@ public class ActiveUserCounter implements AccountDatabaseCrawlerListener {
return tally;
}
private void incrementTallies(String fromNumber, Map<String, long[]> platformIncrements, Map<String, long[]> countryIncrements) {
private void incrementTallies(UUID fromUuid, Map<String, long[]> platformIncrements, Map<String, long[]> countryIncrements) {
try (Jedis jedis = jedisPool.getWriteResource()) {
String tallyValue = jedis.get(TALLY_KEY);
ActiveUserTally activeUserTally;
if (tallyValue == null) {
activeUserTally = new ActiveUserTally(fromNumber, platformIncrements, countryIncrements);
activeUserTally = new ActiveUserTally(fromUuid, platformIncrements, countryIncrements);
} else {
activeUserTally = mapper.readValue(tallyValue, ActiveUserTally.class);
if (activeUserTally.getFromNumber() != fromNumber) {
activeUserTally.setFromNumber(fromNumber);
if (!fromUuid.equals(activeUserTally.getFromUuid())) {
activeUserTally.setFromUuid(fromUuid);
Map<String, long[]> platformTallies = activeUserTally.getPlatforms();
addTallyMaps(platformTallies, platformIncrements);
Map<String, long[]> countryTallies = activeUserTally.getCountries();

35
service/src/main/java/org/whispersystems/textsecuregcm/storage/DirectoryReconciler.java
View File

@@ -30,9 +30,12 @@ import org.whispersystems.textsecuregcm.util.Constants;
import org.whispersystems.textsecuregcm.util.Util;
import javax.ws.rs.ProcessingException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Optional;
import java.util.UUID;
import java.util.stream.Collectors;
import static com.codahale.metrics.MetricRegistry.name;
@@ -55,16 +58,16 @@ public class DirectoryReconciler implements AccountDatabaseCrawlerListener {
public void onCrawlStart() { }
public void onCrawlEnd(Optional<String> fromNumber) {
DirectoryReconciliationRequest request = new DirectoryReconciliationRequest(fromNumber.orElse(null), null, Collections.emptyList());
public void onCrawlEnd(Optional<UUID> fromUuid) {
DirectoryReconciliationRequest request = new DirectoryReconciliationRequest(fromUuid.orElse(null), null, Collections.emptyList());
DirectoryReconciliationResponse response = sendChunk(request);
}
public void onCrawlChunk(Optional<String> fromNumber, List<Account> chunkAccounts) throws AccountDatabaseCrawlerRestartException {
public void onCrawlChunk(Optional<UUID> fromUuid, List<Account> chunkAccounts) throws AccountDatabaseCrawlerRestartException {
updateDirectoryCache(chunkAccounts);
DirectoryReconciliationRequest request = createChunkRequest(fromNumber, chunkAccounts);
DirectoryReconciliationRequest request = createChunkRequest(fromUuid, chunkAccounts);
DirectoryReconciliationResponse response = sendChunk(request);
if (response.getStatus() == DirectoryReconciliationResponse.Status.MISSING) {
throw new AccountDatabaseCrawlerRestartException("directory reconciler missing");
@@ -91,19 +94,21 @@ public class DirectoryReconciler implements AccountDatabaseCrawlerListener {
}
@SuppressWarnings("OptionalUsedAsFieldOrParameterType")
private DirectoryReconciliationRequest createChunkRequest(Optional<String> fromNumber, List<Account> accounts) {
List<String> numbers = accounts.stream()
.filter(Account::isEnabled)
.map(Account::getNumber)
.collect(Collectors.toList());
Optional<String> toNumber = Optional.empty();
if (!accounts.isEmpty()) {
toNumber = Optional.of(accounts.get(accounts.size() - 1).getNumber());
private DirectoryReconciliationRequest createChunkRequest(Optional<UUID> fromUuid, List<Account> accounts) {
List<DirectoryReconciliationRequest.User> users = new ArrayList<>(accounts.size());
for (Account account : accounts) {
if (account.isEnabled()) {
users.add(new DirectoryReconciliationRequest.User(account.getUuid(), account.getNumber()));
}
}
return new DirectoryReconciliationRequest(fromNumber.orElse(null), toNumber.orElse(null), numbers);
Optional<UUID> toUuid = Optional.empty();
if (!accounts.isEmpty()) {
toUuid = Optional.of(accounts.get(accounts.size() - 1).getUuid());
}
return new DirectoryReconciliationRequest(fromUuid.orElse(null), toUuid.orElse(null), users);
}
private DirectoryReconciliationResponse sendChunk(DirectoryReconciliationRequest request) {

2
service/src/main/java/org/whispersystems/textsecuregcm/storage/DirectoryReconciliationClient.java
View File

@@ -51,7 +51,7 @@ public class DirectoryReconciliationClient {
public DirectoryReconciliationResponse sendChunk(DirectoryReconciliationRequest request) {
return client.target(replicationUrl)
.path("/v1/directory/reconcile")
.path("/v2/directory/reconcile")
.request(MediaType.APPLICATION_JSON_TYPE)
.put(Entity.json(request), DirectoryReconciliationResponse.class);
}

5
service/src/main/java/org/whispersystems/textsecuregcm/storage/PushFeedbackProcessor.java
View File

@@ -9,6 +9,7 @@ import org.whispersystems.textsecuregcm.util.Util;
import java.util.List;
import java.util.Optional;
import java.util.UUID;
import java.util.concurrent.TimeUnit;
import static com.codahale.metrics.MetricRegistry.name;
@@ -31,7 +32,7 @@ public class PushFeedbackProcessor implements AccountDatabaseCrawlerListener {
public void onCrawlStart() {}
@Override
public void onCrawlChunk(Optional<String> fromNumber, List<Account> chunkAccounts) {
public void onCrawlChunk(Optional<UUID> fromUuid, List<Account> chunkAccounts) {
for (Account account : chunkAccounts) {
boolean update = false;
@@ -65,5 +66,5 @@ public class PushFeedbackProcessor implements AccountDatabaseCrawlerListener {
}
@Override
public void onCrawlEnd(Optional<String> fromNumber) {}
public void onCrawlEnd(Optional<UUID> toUuid) {}
}