TrueCloudLab/frostfs-sdk-java
17
0
Fork 3
Code Issues 4 Pull requests Releases Packages Activity Actions

[#32] Provide a pool of clients to grpc #33

Merged
orikik merged 1 commit from orikik/frostfs-sdk-java:feature/pool into master 2025-01-17 14:01:11 +00:00
Conversation 0 Commits 1 Files changed 56 +1712 -59
56 changed files with 1712 additions and 59 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files
Showing only changes of commit e9e9480701 - Show all commits

13
client/pom.xml
View file

@ -6,7 +6,7 @@
<parent>
<groupId>info.frostfs.sdk</groupId>
<artifactId>frostfs-sdk-java</artifactId>
<version>0.1.0</version>
<version>0.2.0</version>
</parent>
<artifactId>client</artifactId>
@ -21,17 +21,17 @@
<dependency>
<groupId>info.frostfs.sdk</groupId>
<artifactId>cryptography</artifactId>
<version>0.1.0</version>
<version>0.2.0</version>
</dependency>
<dependency>
<groupId>info.frostfs.sdk</groupId>
<artifactId>models</artifactId>
<version>0.1.0</version>
<version>0.2.0</version>
</dependency>
<dependency>
<groupId>info.frostfs.sdk</groupId>
<artifactId>exceptions</artifactId>
<version>0.1.0</version>
<version>0.2.0</version>
</dependency>
<dependency>
<groupId>commons-codec</groupId>
@ -54,5 +54,10 @@
<artifactId>simpleclient_common</artifactId>
<version>0.16.0</version>
</dependency>
<dependency>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-api</artifactId>
<version>2.0.16</version>
</dependency>
</dependencies>
</project>

60
client/src/main/java/info/frostfs/sdk/FrostFSClient.java
View file

@ -1,5 +1,6 @@
package info.frostfs.sdk;
import frostfs.accounting.Types;
import info.frostfs.sdk.dto.chain.Chain;
import info.frostfs.sdk.dto.chain.ChainTarget;
import info.frostfs.sdk.dto.container.Container;
@ -17,13 +18,16 @@ import info.frostfs.sdk.jdo.ClientEnvironment;
import info.frostfs.sdk.jdo.ClientSettings;
import info.frostfs.sdk.jdo.NetworkSettings;
import info.frostfs.sdk.jdo.PutObjectParameters;
import info.frostfs.sdk.services.*;
import info.frostfs.sdk.pool.SessionCache;
import info.frostfs.sdk.pool.WrapperPrm;
import info.frostfs.sdk.services.CommonClient;
import info.frostfs.sdk.services.impl.*;
import info.frostfs.sdk.services.impl.interceptor.Configuration;
import info.frostfs.sdk.services.impl.interceptor.MonitoringClientInterceptor;
import info.frostfs.sdk.utils.Validator;
import io.grpc.Channel;
import io.grpc.ClientInterceptors;
import io.grpc.ManagedChannel;
import java.util.List;
@ -31,26 +35,28 @@ import static info.frostfs.sdk.constants.ErrorConst.VERSION_UNSUPPORTED_TEMPLATE
import static info.frostfs.sdk.tools.GrpcClient.initGrpcChannel;
import static java.util.Objects.nonNull;
public class FrostFSClient
implements ContainerClient, ObjectClient, ApeManagerClient, NetmapClient, SessionClient, ToolsClient {
public class FrostFSClient implements CommonClient {
private static final MonitoringClientInterceptor MONITORING_CLIENT_INTERCEPTOR =
MonitoringClientInterceptor.create(Configuration.allMetrics());
private final ContainerClientImpl containerClientImpl;
private final ObjectClientImpl objectClientImpl;
private final ApeManagerClientImpl apeManagerClient;
private final NetmapClientImpl netmapClientImpl;
private final SessionClientImpl sessionClientImpl;
private final ObjectToolsImpl objectToolsImpl;
private final AccountingClientImpl accountingClient;
private final ManagedChannel channel;
public FrostFSClient(ClientSettings clientSettings) {
Validator.validate(clientSettings);
Channel channel = nonNull(clientSettings.getChannel())
this.channel = nonNull(clientSettings.getChannel())
? clientSettings.getChannel()
: initGrpcChannel(clientSettings);
MonitoringClientInterceptor monitoringClientInterceptor = MonitoringClientInterceptor
.create(Configuration.allMetrics());
channel = ClientInterceptors.intercept(channel, monitoringClientInterceptor);
Channel interceptChannel = ClientInterceptors.intercept(channel, MONITORING_CLIENT_INTERCEPTOR);
ClientEnvironment clientEnvironment =
new ClientEnvironment(clientSettings.getKey(), channel, new Version(), this);
new ClientEnvironment(clientSettings.getKey(), interceptChannel, new Version(), this);
Validator.validate(clientEnvironment);
@ -60,10 +66,30 @@ public class FrostFSClient
this.netmapClientImpl = new NetmapClientImpl(clientEnvironment);
this.sessionClientImpl = new SessionClientImpl(clientEnvironment);
this.objectToolsImpl = new ObjectToolsImpl(clientEnvironment);
checkFrostFsVersionSupport(clientEnvironment.getVersion());
this.accountingClient = new AccountingClientImpl(clientEnvironment);
checkFrostFSVersionSupport(clientEnvironment.getVersion());
}
private void checkFrostFsVersionSupport(Version version) {
public FrostFSClient(WrapperPrm prm, SessionCache cache) {
this.channel = initGrpcChannel(prm.getAddress());
Channel interceptChannel = ClientInterceptors.intercept(channel, MONITORING_CLIENT_INTERCEPTOR);
ClientEnvironment clientEnvironment =
new ClientEnvironment(prm.getKey(), interceptChannel, new Version(), this);
Validator.validate(clientEnvironment);
this.containerClientImpl = new ContainerClientImpl(clientEnvironment);
this.objectClientImpl = new ObjectClientImpl(clientEnvironment);
this.apeManagerClient = new ApeManagerClientImpl(clientEnvironment);
this.netmapClientImpl = new NetmapClientImpl(clientEnvironment);
this.sessionClientImpl = new SessionClientImpl(clientEnvironment);
this.objectToolsImpl = new ObjectToolsImpl(clientEnvironment);
this.accountingClient = new AccountingClientImpl(clientEnvironment);
checkFrostFSVersionSupport(clientEnvironment.getVersion());
}
private void checkFrostFSVersionSupport(Version version) {
var localNodeInfo = netmapClientImpl.getLocalNodeInfo();
if (!localNodeInfo.getVersion().isSupported(version)) {
throw new ProcessFrostFSException(
@ -165,4 +191,18 @@ public class FrostFSClient
public ObjectId calculateObjectId(ObjectHeader header) {
return objectToolsImpl.calculateObjectId(header);
}
@Override
public Types.Decimal getBalance() {
return accountingClient.getBalance();
}
public String dial() {
accountingClient.getBalance();
return null;
}
public void close() {
channel.shutdown();
}
}

14
client/src/main/java/info/frostfs/sdk/constants/PoolConst.java Normal file
View file

@ -0,0 +1,14 @@
package info.frostfs.sdk.constants;
public class PoolConst {
public static final int DEFAULT_SESSION_TOKEN_EXPIRATION_DURATION = 100; // in epochs
public static final int DEFAULT_ERROR_THRESHOLD = 100;
public static final int DEFAULT_GRACEFUL_CLOSE_ON_SWITCH_TIMEOUT = 10; // Seconds
public static final int DEFAULT_REBALANCE_INTERVAL = 15; // Seconds
public static final int DEFAULT_HEALTHCHECK_TIMEOUT = 4; // Seconds
public static final int DEFAULT_DIAL_TIMEOUT = 5; // Seconds
public static final int DEFAULT_STREAM_TIMEOUT = 10; // Seconds
private PoolConst() {
}
}

23
client/src/main/java/info/frostfs/sdk/enums/HealthyStatus.java Normal file
View file

@ -0,0 +1,23 @@
package info.frostfs.sdk.enums;
public enum HealthyStatus {
// status HEALTHY is set when connection is ready to be used by the pool.
HEALTHY(1),
// status UNHEALTHY_ON_REQUEST is set when communication after dialing to the
// endpoint is failed due to immediate or accumulated errors, connection is
// available and pool should close it before re-establishing connection once again.
UNHEALTHY_ON_REQUEST(2),
// status UNHEALTHY_ON_DIAL is set when dialing to the endpoint is failed,
// so there is no connection to the endpoint, and pool should not close it
// before re-establishing connection once again.
UNHEALTHY_ON_DIAL(3),
;
public final int value;
HealthyStatus(int value) {
this.value = value;
}
}

29
client/src/main/java/info/frostfs/sdk/enums/MethodIndex.java Normal file
View file

@ -0,0 +1,29 @@
package info.frostfs.sdk.enums;
public enum MethodIndex {
METHOD_BALANCE_GET("balanceGet"),
METHOD_CONTAINER_PUT("containerPut"),
METHOD_CONTAINER_GET("ContainerGet"),
METHOD_CONTAINER_LIST("ContainerList"),
METHOD_CONTAINER_DELETE("ContainerDelete"),
METHOD_ENDPOINT_INFO("EndpointInfo"),
METHOD_NETWORK_INFO("NetworkInfo"),
METHOD_NETMAP_SNAPSHOT("NetMapSnapshot"),
METHOD_OBJECT_PUT("ObjectPut"),
METHOD_OBJECT_DELETE("ObjectDelete"),
METHOD_OBJECT_GET("ObjectGet"),
METHOD_OBJECT_HEAD("ObjectHead"),
METHOD_OBJECT_RANGE("ObjectRange"),
METHOD_OBJECT_PATCH("ObjectPatch"),
METHOD_SESSION_CREATE("SessionCreate"),
METHOD_APE_MANAGER_ADD_CHAIN("APEManagerAddChain"),
METHOD_APE_MANAGER_REMOVE_CHAIN("APEManagerRemoveChain"),
METHOD_APE_MANAGER_LIST_CHAINS("APEManagerListChains"),
;
public final String methodName;
MethodIndex(String methodName) {
this.methodName = methodName;
}
}

2
client/src/main/java/info/frostfs/sdk/exceptions/ResponseFrostFSException.java
View file

@ -4,7 +4,7 @@ import info.frostfs.sdk.dto.response.ResponseStatus;
import lombok.Getter;
@Getter
public class ResponseFrostFSException extends RuntimeException {
public class ResponseFrostFSException extends FrostFSException {
private final ResponseStatus status;
public ResponseFrostFSException(ResponseStatus status) {

8
client/src/main/java/info/frostfs/sdk/jdo/ClientEnvironment.java
View file

@ -38,4 +38,12 @@ public class ClientEnvironment {
this.channel = channel;
this.frostFSClient = frostFSClient;
}
public ClientEnvironment(ECDsa key, Channel channel, Version version, FrostFSClient frostFSClient) {
this.key = key;
this.ownerId = new OwnerId(key.getPublicKeyByte());
this.version = version;
this.channel = channel;
this.frostFSClient = frostFSClient;
}
}

6
client/src/main/java/info/frostfs/sdk/jdo/ClientSettings.java
View file

@ -2,8 +2,8 @@ package info.frostfs.sdk.jdo;
import info.frostfs.sdk.annotations.AtLeastOneIsFilled;
import info.frostfs.sdk.annotations.NotNull;
import io.grpc.Channel;
import io.grpc.ChannelCredentials;
import io.grpc.ManagedChannel;
import lombok.Getter;
import lombok.experimental.FieldNameConstants;
@ -17,7 +17,7 @@ public class ClientSettings {
private String host;
private ChannelCredentials credentials;
private Channel channel;
private ManagedChannel channel;
public ClientSettings(String key, String host) {
this.key = key;
@ -30,7 +30,7 @@ public class ClientSettings {
this.credentials = credentials;
}
public ClientSettings(String key, Channel channel) {
public ClientSettings(String key, ManagedChannel channel) {
this.key = key;
this.channel = channel;
}

12
client/src/main/java/info/frostfs/sdk/jdo/pool/NodeParameters.java Normal file
View file

@ -0,0 +1,12 @@
package info.frostfs.sdk.jdo.pool;
import lombok.AllArgsConstructor;
import lombok.Getter;
@Getter
@AllArgsConstructor
public class NodeParameters {
private final int priority;
private final String address;
private final double weight;
}

44
client/src/main/java/info/frostfs/sdk/jdo/pool/PoolInitParameters.java Normal file
View file

@ -0,0 +1,44 @@
package info.frostfs.sdk.jdo.pool;
import info.frostfs.sdk.jdo.ECDsa;
import info.frostfs.sdk.pool.ClientWrapper;
import io.grpc.ClientInterceptors;
import io.netty.channel.ChannelOption;
import lombok.Getter;
import lombok.Setter;
import org.slf4j.Logger;
import java.util.ArrayList;
import java.util.Collection;
import java.util.function.Function;
@Getter
@Setter
public class PoolInitParameters {
private ECDsa key;
private long nodeDialTimeout;
private long nodeStreamTimeout;
private long healthCheckTimeout;
private long clientRebalanceInterval;
private long sessionExpirationDuration;
private int errorThreshold;
private NodeParameters[] nodeParams;
private ChannelOption<?>[] dialOptions;
private Function<String, ClientWrapper> clientBuilder;
private long gracefulCloseOnSwitchTimeout;
private Logger logger;
private Collection<ClientInterceptors> interceptors = new ArrayList<>();
}

27
client/src/main/java/info/frostfs/sdk/pool/ClientStatus.java Normal file
View file

@ -0,0 +1,27 @@
package info.frostfs.sdk.pool;
public interface ClientStatus {
// isHealthy checks if the connection can handle requests.
boolean isHealthy();
// isDialed checks if the connection was created.
boolean isDialed();
// setUnhealthy marks client as unhealthy.
void setUnhealthy();
// address return address of endpoint.
String getAddress();
// currentErrorRate returns current errors rate.
// After specific threshold connection is considered as unhealthy.
// Pool.startRebalance routine can make this connection healthy again.
int getCurrentErrorRate();
// overallErrorRate returns the number of all happened errors.
long getOverallErrorRate();
// methodsStatus returns statistic for all used methods.
StatusSnapshot[] getMethodsStatus();
}

114
client/src/main/java/info/frostfs/sdk/pool/ClientStatusMonitor.java Normal file
View file

@ -0,0 +1,114 @@
package info.frostfs.sdk.pool;
import info.frostfs.sdk.enums.HealthyStatus;
import info.frostfs.sdk.enums.MethodIndex;
import info.frostfs.sdk.utils.FrostFSMessages;
import lombok.Getter;
import lombok.Setter;
import org.slf4j.Logger;
import java.util.Arrays;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.ReentrantLock;
@Getter
@Setter
public class ClientStatusMonitor implements ClientStatus {
private final ReentrantLock lock = new ReentrantLock();
private final Logger logger;
private final AtomicInteger healthy = new AtomicInteger();
private final String address;
private final MethodStatus[] methods;
private int errorThreshold;
private int currentErrorCount;
private long overallErrorCount;
public ClientStatusMonitor(Logger logger, String address) {
this.logger = logger;
this.healthy.set(HealthyStatus.HEALTHY.value);
this.address = address;
this.methods = Arrays.stream(MethodIndex.values())
.map(t -> new MethodStatus(t.methodName))
.toArray(MethodStatus[]::new);
}
@Override
public boolean isHealthy() {
return healthy.get() == HealthyStatus.HEALTHY.value;
}
@Override
public boolean isDialed() {
return healthy.get() != HealthyStatus.UNHEALTHY_ON_DIAL.value;
}
public void setHealthy() {
healthy.set(HealthyStatus.HEALTHY.ordinal());
}
@Override
public void setUnhealthy() {
healthy.set(HealthyStatus.UNHEALTHY_ON_REQUEST.value);
}
public void setUnhealthyOnDial() {
healthy.set(HealthyStatus.UNHEALTHY_ON_DIAL.value);
}
public void incErrorRate() {
boolean thresholdReached;
lock.lock();
try {
currentErrorCount++;
overallErrorCount++;
thresholdReached = currentErrorCount >= errorThreshold;
if (thresholdReached) {
setUnhealthy();
currentErrorCount = 0;
}
} finally {
lock.unlock();
}
if (thresholdReached && logger != null) {
FrostFSMessages.errorThresholdReached(logger, address, errorThreshold);
}
}
@Override
public int getCurrentErrorRate() {
lock.lock();
try {
return currentErrorCount;
} finally {
lock.unlock();
}
}
@Override
public long getOverallErrorRate() {
lock.lock();
try {
return overallErrorCount;
} finally {
lock.unlock();
}
}
@Override
public StatusSnapshot[] getMethodsStatus() {
StatusSnapshot[] result = new StatusSnapshot[methods.length];
for (int i = 0; i < result.length; i++) {
result[i] = methods[i].getSnapshot();
}
return result;
}
}

130
client/src/main/java/info/frostfs/sdk/pool/ClientWrapper.java Normal file
View file

@ -0,0 +1,130 @@
package info.frostfs.sdk.pool;
import info.frostfs.sdk.FrostFSClient;
import info.frostfs.sdk.enums.MethodIndex;
import info.frostfs.sdk.exceptions.ResponseFrostFSException;
import info.frostfs.sdk.exceptions.ValidationFrostFSException;
import info.frostfs.sdk.utils.WaitUtil;
import lombok.AccessLevel;
import lombok.Getter;
import org.apache.commons.lang3.StringUtils;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;
import static info.frostfs.sdk.constants.ErrorConst.POOL_CLIENT_UNHEALTHY;
@Getter
public class ClientWrapper extends ClientStatusMonitor {
@Getter(value = AccessLevel.NONE)
private final Lock lock = new ReentrantLock();
private final SessionCache sessionCache;
private final WrapperPrm wrapperPrm;
private FrostFSClient client;
public ClientWrapper(WrapperPrm wrapperPrm, Pool pool) {
super(wrapperPrm.getLogger(), wrapperPrm.getAddress());
this.wrapperPrm = wrapperPrm;
setErrorThreshold(wrapperPrm.getErrorThreshold());
this.sessionCache = pool.getSessionCache();
this.client = new FrostFSClient(wrapperPrm, sessionCache);
}
public FrostFSClient getClient() {
lock.lock();
try {
if (isHealthy()) {
return client;
}
return null;
} finally {
lock.unlock();
}
}
public void dial() {
FrostFSClient client = getClient();
if (client == null) {
throw new ValidationFrostFSException(POOL_CLIENT_UNHEALTHY);
}
client.dial();
}
public void handleError(Exception exp) {
if (exp instanceof ResponseFrostFSException && ((ResponseFrostFSException) exp).getStatus() != null) {
switch (((ResponseFrostFSException) exp).getStatus().getCode()) {
case INTERNAL:
case WRONG_MAGIC_NUMBER:
case SIGNATURE_VERIFICATION_FAILURE:
case NODE_UNDER_MAINTENANCE:
incErrorRate();
}
return;
}
incErrorRate();
}
private void scheduleGracefulClose() {
if (client == null) {
return;
}
WaitUtil.sleep(wrapperPrm.getGracefulCloseOnSwitchTimeout());
client.close();
}
public CompletableFuture<Boolean> restartIfUnhealthy() {
try {
client.getLocalNodeInfo();
return CompletableFuture.completedFuture(false);
} catch (Exception ignored) {
}
if (isDialed()) {
scheduleGracefulClose();
}
return CompletableFuture.completedFuture(restartClient());
}
private boolean restartClient() {
FrostFSClient newClient = null;
try {
newClient = new FrostFSClient(wrapperPrm, sessionCache);
var error = newClient.dial();
if (StringUtils.isNotBlank(error)) {
setUnhealthyOnDial();
newClient.close();
return true;
}
lock.lock();
client = newClient;
lock.unlock();
} catch (Exception exp) {
if (newClient != null) {
newClient.close();
}
}
try {
client.getLocalNodeInfo();
} catch (Exception exp) {
setUnhealthy();
return true;
}
setHealthy();
return false;
}
public void incRequests(long elapsed, MethodIndex method) {
var methodStat = getMethods()[method.ordinal()];
methodStat.incRequests(elapsed);
}
}

54
client/src/main/java/info/frostfs/sdk/pool/InnerPool.java Normal file
View file

@ -0,0 +1,54 @@
package info.frostfs.sdk.pool;
import java.util.concurrent.locks.ReentrantLock;
class InnerPool {
private static final int ATTEMPTS_COUNT = 3;
private final ReentrantLock lock = new ReentrantLock();
private final ClientWrapper[] clients;
private Sampler sampler;
InnerPool(Sampler sampler, ClientWrapper[] clients) {
this.sampler = sampler;
this.clients = clients;
}
Sampler getSampler() {
return sampler;
}
void setSampler(Sampler sampler) {
this.sampler = sampler;
}
ClientWrapper[] getClients() {
return clients;
}
ClientWrapper connection() {
lock.lock();
try {
if (clients.length == 1) {
ClientWrapper client = clients[0];
if (client.isHealthy()) {
return client;
}
} else {
int attempts = ATTEMPTS_COUNT * clients.length;
for (int i = 0; i < attempts; i++) {
int index = sampler.next();
if (clients[index].isHealthy()) {
return clients[index];
}
}
}
return null;
} finally {
lock.unlock();
}
}
}

31
client/src/main/java/info/frostfs/sdk/pool/MethodStatus.java Normal file
View file

@ -0,0 +1,31 @@
package info.frostfs.sdk.pool;
import lombok.AccessLevel;
import lombok.Getter;
import lombok.Setter;
import java.util.concurrent.locks.ReentrantLock;
@Setter
@Getter
public class MethodStatus {
@Getter(AccessLevel.NONE)
private final ReentrantLock lock = new ReentrantLock();
private final String name;
private StatusSnapshot snapshot;
public MethodStatus(String name) {
this.name = name;
this.snapshot = new StatusSnapshot();
}
void incRequests(long elapsed) {
lock.lock();
try {
snapshot.setAllTime(snapshot.getAllTime() + elapsed);
snapshot.setAllRequests(snapshot.getAllRequests() + 1);
} finally {
lock.unlock();
}
}
}

13
client/src/main/java/info/frostfs/sdk/pool/NodeStatistic.java Normal file
View file

@ -0,0 +1,13 @@
package info.frostfs.sdk.pool;
import lombok.Getter;
import lombok.Setter;
@Getter
@Setter
public class NodeStatistic {
private String address;
private StatusSnapshot[] methods;
private long overallErrors;
private int currentErrors;
}

19
client/src/main/java/info/frostfs/sdk/pool/NodesParam.java Normal file
View file

@ -0,0 +1,19 @@
package info.frostfs.sdk.pool;
import lombok.Getter;
import java.util.ArrayList;
import java.util.List;
@Getter
public class NodesParam {
private final int priority;
private final List<String> address;
private final List<Double> weight;
public NodesParam(int priority) {
this.priority = priority;
this.address = new ArrayList<>();
this.weight = new ArrayList<>();
}
}

520
client/src/main/java/info/frostfs/sdk/pool/Pool.java Normal file
View file

@ -0,0 +1,520 @@
package info.frostfs.sdk.pool;
import frostfs.refs.Types;
import info.frostfs.sdk.dto.chain.Chain;
import info.frostfs.sdk.dto.chain.ChainTarget;
import info.frostfs.sdk.dto.container.Container;
import info.frostfs.sdk.dto.container.ContainerId;
import info.frostfs.sdk.dto.netmap.NetmapSnapshot;
import info.frostfs.sdk.dto.netmap.NodeInfo;
import info.frostfs.sdk.dto.object.*;
import info.frostfs.sdk.dto.session.SessionToken;
import info.frostfs.sdk.exceptions.FrostFSException;
import info.frostfs.sdk.exceptions.SessionExpiredFrostFSException;
import info.frostfs.sdk.exceptions.SessionNotFoundFrostFSException;
import info.frostfs.sdk.exceptions.ValidationFrostFSException;
import info.frostfs.sdk.jdo.ECDsa;
import info.frostfs.sdk.jdo.NetworkSettings;
import info.frostfs.sdk.jdo.PutObjectParameters;
import info.frostfs.sdk.jdo.pool.NodeParameters;
import info.frostfs.sdk.jdo.pool.PoolInitParameters;
import info.frostfs.sdk.services.CommonClient;
import info.frostfs.sdk.utils.FrostFSMessages;
import info.frostfs.sdk.utils.WaitUtil;
import lombok.Getter;
import org.slf4j.Logger;
import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.locks.ReentrantLock;
import java.util.function.Function;
import static info.frostfs.sdk.Helper.getHexString;
import static info.frostfs.sdk.constants.ErrorConst.*;
import static info.frostfs.sdk.constants.PoolConst.*;
@Getter
public class Pool implements CommonClient {
private final ReentrantLock lock = new ReentrantLock();
private final ECDsa key;
private final SessionCache sessionCache;
private final long sessionTokenDuration;
private final RebalanceParameters rebalanceParams;
private final Function<String, ClientWrapper> clientBuilder;
private final Logger logger;
private InnerPool[] innerPools;
private Types.OwnerID ownerID;
private OwnerId ownerId;
private boolean disposedValue;
private long maxObjectSize;
private ClientStatus clientStatus;
public Pool(PoolInitParameters options) {
if (options == null || options.getKey() == null) {
throw new ValidationFrostFSException(
String.format(
PARAMS_ARE_MISSING_TEMPLATE,
String.join(
FIELDS_DELIMITER_COMMA, PoolInitParameters.class.getName(), ECDsa.class.getName()
)
)
);
}
List<NodesParam> nodesParams = adjustNodeParams(options.getNodeParams());
SessionCache cache = new SessionCache(options.getSessionExpirationDuration());
fillDefaultInitParams(options, this);
this.key = options.getKey();
this.sessionCache = cache;
this.logger = options.getLogger();
this.sessionTokenDuration = options.getSessionExpirationDuration();
this.rebalanceParams = new RebalanceParameters(
nodesParams.toArray(new NodesParam[0]),
options.getHealthCheckTimeout(),
options.getClientRebalanceInterval(),
options.getSessionExpirationDuration());
this.clientBuilder = options.getClientBuilder();
}
private static List<NodesParam> adjustNodeParams(NodeParameters[] nodeParams) {
if (nodeParams == null || nodeParams.length == 0) {
throw new ValidationFrostFSException(POOL_PEERS_IS_MISSING);
}
Map<Integer, NodesParam> nodesParamsDict = new HashMap<>(nodeParams.length);
for (NodeParameters nodeParam : nodeParams) {
var nodesParam = nodesParamsDict
.computeIfAbsent(nodeParam.getPriority(), k -> new NodesParam(nodeParam.getPriority()));
nodesParam.getAddress().add(nodeParam.getAddress());
nodesParam.getWeight().add(nodeParam.getWeight());
}
List<NodesParam> nodesParams = new ArrayList<>(nodesParamsDict.values());
nodesParams.sort(Comparator.comparingInt(NodesParam::getPriority));
for (NodesParam nodes : nodesParams) {
double[] newWeights = adjustWeights(nodes.getWeight().stream().mapToDouble(Double::doubleValue).toArray());
nodes.getWeight().clear();
for (double weight : newWeights) {
nodes.getWeight().add(weight);
}
}
return nodesParams;
}
private static double[] adjustWeights(double[] weights) {
double[] adjusted = new double[weights.length];
double sum = Arrays.stream(weights).sum();
if (sum > 0) {
for (int i = 0; i < weights.length; i++) {
adjusted[i] = weights[i] / sum;
}
}
return adjusted;
}
private static void fillDefaultInitParams(PoolInitParameters parameters, Pool pool) {
if (parameters.getSessionExpirationDuration() == 0) {
parameters.setSessionExpirationDuration(DEFAULT_SESSION_TOKEN_EXPIRATION_DURATION);
}
if (parameters.getErrorThreshold() == 0) {
parameters.setErrorThreshold(DEFAULT_ERROR_THRESHOLD);
}
if (parameters.getClientRebalanceInterval() <= 0) {
parameters.setClientRebalanceInterval(DEFAULT_REBALANCE_INTERVAL);
}
if (parameters.getGracefulCloseOnSwitchTimeout() <= 0) {
parameters.setGracefulCloseOnSwitchTimeout(DEFAULT_GRACEFUL_CLOSE_ON_SWITCH_TIMEOUT);
}
if (parameters.getHealthCheckTimeout() <= 0) {
parameters.setHealthCheckTimeout(DEFAULT_HEALTHCHECK_TIMEOUT);
}
if (parameters.getNodeDialTimeout() <= 0) {
parameters.setNodeDialTimeout(DEFAULT_DIAL_TIMEOUT);
}
if (parameters.getNodeStreamTimeout() <= 0) {
parameters.setNodeStreamTimeout(DEFAULT_STREAM_TIMEOUT);
}
if (parameters.getSessionExpirationDuration() == 0) {
parameters.setSessionExpirationDuration(DEFAULT_SESSION_TOKEN_EXPIRATION_DURATION);
}
if (parameters.getClientBuilder() == null) {
parameters.setClientBuilder(address -> {
WrapperPrm wrapperPrm = new WrapperPrm();
wrapperPrm.setAddress(address);
wrapperPrm.setKey(parameters.getKey());
wrapperPrm.setLogger(parameters.getLogger());
wrapperPrm.setDialTimeout(parameters.getNodeDialTimeout());
wrapperPrm.setStreamTimeout(parameters.getNodeStreamTimeout());
wrapperPrm.setErrorThreshold(parameters.getErrorThreshold());
wrapperPrm.setGracefulCloseOnSwitchTimeout(parameters.getGracefulCloseOnSwitchTimeout());
wrapperPrm.setInterceptors(parameters.getInterceptors());
return new ClientWrapper(wrapperPrm, pool);
});
}
}
private static SessionToken initSessionForDuration(ClientWrapper cw, long duration) {
var client = cw.getClient();
NetworkSettings networkInfo = client.getNetworkSettings();
long epoch = networkInfo.getEpochDuration();
long exp = (Long.MAX_VALUE - epoch < duration) ? Long.MAX_VALUE : (epoch + duration);
return client.createSession(exp);
}
private static String formCacheKey(String address, String key) {
return address + key;
}
public String dial() {
InnerPool[] inner = new InnerPool[rebalanceParams.getNodesParams().length];
boolean atLeastOneHealthy = false;
int i = 0;
for (NodesParam nodeParams : rebalanceParams.getNodesParams()) {
ClientWrapper[] clients = new ClientWrapper[nodeParams.getWeight().size()];
for (int j = 0; j < nodeParams.getAddress().size(); j++) {
ClientWrapper client = clients[j] = clientBuilder.apply(nodeParams.getAddress().get(j));
boolean dialed = false;
try {
client.dial();
dialed = true;
SessionToken token = initSessionForDuration(client, rebalanceParams.getSessionExpirationDuration());
String cacheKey = formCacheKey(
nodeParams.getAddress().get(j),
getHexString(key.getPublicKeyByte())
);
sessionCache.setValue(cacheKey, token);
atLeastOneHealthy = true;
} catch (ValidationFrostFSException exp) {
break;
} catch (Exception exp) {
if (!dialed) {
client.setUnhealthyOnDial();
} else {
client.setUnhealthy();
}
if (logger != null) {
FrostFSMessages
.sessionCreationError(logger, client.getWrapperPrm().getAddress(), exp.getMessage());
}
}
}
Sampler sampler = new Sampler(nodeParams.getWeight().stream().mapToDouble(Double::doubleValue).toArray());
inner[i] = new InnerPool(sampler, clients);
i++;
}
if (!atLeastOneHealthy) {
return POOL_NODES_UNHEALTHY;
}
this.innerPools = inner;
NetworkSettings networkSettings = getNetworkSettings();
this.maxObjectSize = networkSettings.getMaxObjectSize();
startRebalance();
return null;
}
private ClientWrapper connection() {
for (InnerPool pool : innerPools) {
ClientWrapper client = pool.connection();
if (client != null) {
return client;
}
}
throw new FrostFSException(POOL_CLIENTS_UNHEALTHY);
}
public void close() {
if (innerPools != null) {
for (InnerPool innerPool : innerPools) {
for (ClientWrapper client : innerPool.getClients()) {
if (client.isDialed()) {
client.getClient().close();
}
}
}
}
}
public void startRebalance() {
double[][] buffers = new double[rebalanceParams.getNodesParams().length][];
for (int i = 0; i < rebalanceParams.getNodesParams().length; i++) {
NodesParam parameters = rebalanceParams.getNodesParams()[i];
buffers[i] = new double[parameters.getWeight().size()];
CompletableFuture.runAsync(() -> {
WaitUtil.sleep(rebalanceParams.getClientRebalanceInterval());
updateNodesHealth(buffers);
});
}
}
private void updateNodesHealth(double[][] buffers) {
CompletableFuture<?>[] tasks = new CompletableFuture<?>[innerPools.length];
for (int i = 0; i < innerPools.length; i++) {
double[] bufferWeights = buffers[i];
int finalI = i;
tasks[i] = CompletableFuture.runAsync(() -> updateInnerNodesHealth(finalI, bufferWeights));
}
CompletableFuture.allOf(tasks).join();
}
private void updateInnerNodesHealth(int poolIndex, double[] bufferWeights) {
if (poolIndex > innerPools.length - 1) {
return;
}
InnerPool pool = innerPools[poolIndex];
RebalanceParameters options = rebalanceParams;
int[] healthyChanged = {0};
CompletableFuture<?>[] tasks = new CompletableFuture<?>[pool.getClients().length];
for (int j = 0; j < pool.getClients().length; j++) {
ClientWrapper client = pool.getClients()[j];
AtomicBoolean healthy = new AtomicBoolean(false);
AtomicReference<String> error = new AtomicReference<>();
AtomicBoolean changed = new AtomicBoolean(false);
int finalJ = j;
tasks[j] = client.restartIfUnhealthy().handle((unused, throwable) -> {
if (throwable != null) {
error.set(throwable.getMessage());
bufferWeights[finalJ] = 0;
sessionCache.deleteByPrefix(client.getAddress());
} else {
changed.set(unused);
healthy.set(true);
bufferWeights[finalJ] = options.getNodesParams()[poolIndex].getWeight().get(finalJ);
}
return null;
}).thenRun(() -> {
if (changed.get()) {
if (error.get() != null && logger != null) {
FrostFSMessages.healthChanged(logger, client.getAddress(), healthy.get(), error.get());
}
healthyChanged[0] = 1;
}
});
}
CompletableFuture.allOf(tasks).thenRun(() -> {
if (healthyChanged[0] == 1) {
double[] probabilities = adjustWeights(bufferWeights);
lock.lock();
try {
pool.setSampler(new Sampler(probabilities));
} finally {
lock.unlock();
}
}
});
}
private boolean checkSessionTokenErr(Exception error, String address) {
if (error == null) {
return false;
}
if (error instanceof SessionNotFoundFrostFSException || error instanceof SessionExpiredFrostFSException) {
sessionCache.deleteByPrefix(address);
return true;
}
return false;
}
public Statistic statistic() {
if (innerPools == null) {
throw new ValidationFrostFSException(POOL_NOT_DIALED);
}
Statistic statistics = new Statistic();
for (InnerPool inner : innerPools) {
int valueIndex = 0;
String[] nodes = new String[inner.getClients().length];
lock.lock();
try {
for (ClientWrapper client : inner.getClients()) {
if (client.isHealthy()) {
nodes[valueIndex] = client.getAddress();
}
NodeStatistic node = new NodeStatistic();
node.setAddress(client.getAddress());
node.setMethods(client.getMethodsStatus());
node.setOverallErrors(client.getOverallErrorRate());
node.setCurrentErrors(client.getCurrentErrorRate());
statistics.getNodes().add(node);
valueIndex++;
statistics.setOverallErrors(statistics.getOverallErrors() + node.getOverallErrors());
}
if (statistics.getCurrentNodes() == null || statistics.getCurrentNodes().length == 0) {
statistics.setCurrentNodes(nodes);
}
} finally {
lock.unlock();
}
}
return statistics;
}
@Override
public Container getContainer(ContainerId cid) {
ClientWrapper client = connection();
return client.getClient().getContainer(cid);
}
@Override
public List<ContainerId> listContainers() {
ClientWrapper client = connection();
return client.getClient().listContainers();
}
@Override
public ContainerId createContainer(Container container) {
ClientWrapper client = connection();
return client.getClient().createContainer(container);
}
@Override
public void deleteContainer(ContainerId cid) {
ClientWrapper client = connection();
client.getClient().deleteContainer(cid);
}
@Override
public ObjectHeader getObjectHead(ContainerId containerId, ObjectId objectId) {
ClientWrapper client = connection();
return client.getClient().getObjectHead(containerId, objectId);
}
@Override
public ObjectFrostFS getObject(ContainerId containerId, ObjectId objectId) {
ClientWrapper client = connection();
return client.getClient().getObject(containerId, objectId);
}
@Override
public ObjectId putObject(PutObjectParameters parameters) {
ClientWrapper client = connection();
return client.getClient().putObject(parameters);
}
@Override
public ObjectId putSingleObject(ObjectFrostFS objectFrostFS) {
ClientWrapper client = connection();
return client.getClient().putSingleObject(objectFrostFS);
}
@Override
public void deleteObject(ContainerId containerId, ObjectId objectId) {
ClientWrapper client = connection();
client.getClient().deleteObject(containerId, objectId);
}
@Override
public Iterable<ObjectId> searchObjects(ContainerId cid, ObjectFilter<?>... filters) {
ClientWrapper client = connection();
return client.getClient().searchObjects(cid, filters);
}
@Override
public byte[] addChain(Chain chain, ChainTarget chainTarget) {
ClientWrapper client = connection();
return client.getClient().addChain(chain, chainTarget);
}
@Override
public void removeChain(Chain chain, ChainTarget chainTarget) {
ClientWrapper client = connection();
client.getClient().removeChain(chain, chainTarget);
}
@Override
public List<Chain> listChains(ChainTarget chainTarget) {
ClientWrapper client = connection();
return client.getClient().listChains(chainTarget);
}
@Override
public NetmapSnapshot getNetmapSnapshot() {
ClientWrapper client = connection();
return client.getClient().getNetmapSnapshot();
}
@Override
public NodeInfo getLocalNodeInfo() {
ClientWrapper client = connection();
return client.getClient().getLocalNodeInfo();
}
@Override
public NetworkSettings getNetworkSettings() {
ClientWrapper client = connection();
return client.getClient().getNetworkSettings();
}
@Override
public SessionToken createSession(long expiration) {
ClientWrapper client = connection();
return client.getClient().createSession(expiration);
}
public frostfs.session.Types.SessionToken createSessionInternal(long expiration) {
ClientWrapper client = connection();
return client.getClient().createSessionInternal(expiration);
}
@Override
public ObjectId calculateObjectId(ObjectHeader header) {
ClientWrapper client = connection();
return client.getClient().calculateObjectId(header);
}
@Override
public frostfs.accounting.Types.Decimal getBalance() {
ClientWrapper client = connection();
return client.getClient().getBalance();
}
}

15
client/src/main/java/info/frostfs/sdk/pool/RebalanceParameters.java Normal file
View file

@ -0,0 +1,15 @@
package info.frostfs.sdk.pool;
import lombok.AllArgsConstructor;
import lombok.Getter;
import lombok.Setter;
@Getter
@Setter
@AllArgsConstructor
public class RebalanceParameters {
private NodesParam[] nodesParams;
private long nodeRequestTimeout;
private long clientRebalanceInterval;
private long sessionExpirationDuration;
}

15
client/src/main/java/info/frostfs/sdk/pool/RequestInfo.java Normal file
View file

@ -0,0 +1,15 @@
package info.frostfs.sdk.pool;
import info.frostfs.sdk.enums.MethodIndex;
import lombok.Getter;
import lombok.Setter;
import java.time.Duration;
@Getter
@Setter
public class RequestInfo {
private String address;
private MethodIndex methodIndex;
private Duration elapsed;
}

77
client/src/main/java/info/frostfs/sdk/pool/Sampler.java Normal file
View file

@ -0,0 +1,77 @@
package info.frostfs.sdk.pool;
import java.util.ArrayList;
import java.util.Random;
class Sampler {
private final Object lock = new Object();
private final Random random = new Random();
private final double[] probabilities;
private final int[] alias;
Sampler(double[] probabilities) {
ArrayList<Integer> small = new ArrayList<>();
ArrayList<Integer> large = new ArrayList<>();
int n = probabilities.length;
this.probabilities = new double[n];
this.alias = new int[n];
// Compute scaled probabilities.
double[] p = new double[n];
for (int i = 0; i < n; i++) {
p[i] = probabilities[i] * n;
if (p[i] < 1) {
small.add(i);
} else {
large.add(i);
}
}
while (!small.isEmpty() && !large.isEmpty()) {
int l = small.remove(small.size() - 1);
int g = large.remove(large.size() - 1);
this.probabilities[l] = p[l];
this.alias[l] = g;
p[g] = p[g] + p[l] - 1;
if (p[g] < 1) {
small.add(g);
} else {
large.add(g);
}
}
while (!large.isEmpty()) {
int g = large.remove(large.size() - 1);
this.probabilities[g] = 1;
}
while (!small.isEmpty()) {
int l = small.remove(small.size() - 1);
probabilities[l] = 1;
}
}
int next() {
int n = alias.length;
int i;
double f;
synchronized (lock) {
i = random.nextInt(n);
f = random.nextDouble();
}
if (f < probabilities[i]) {
return i;
}
return alias[i];
}
}

46
client/src/main/java/info/frostfs/sdk/pool/SessionCache.java Normal file
View file

@ -0,0 +1,46 @@
package info.frostfs.sdk.pool;
import info.frostfs.sdk.dto.session.SessionToken;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
public class SessionCache {
private final ConcurrentMap<String, SessionToken> cache = new ConcurrentHashMap<>();
private final long tokenDuration;
private long currentEpoch;
public SessionCache(long sessionExpirationDuration) {
this.tokenDuration = sessionExpirationDuration;
}
public boolean contains(String key) {
return cache.containsKey(key);
}
public boolean tryGetValue(String key, SessionToken[] value) {
if (key == null) {
value[0] = null;
return false;
}
SessionToken token = cache.get(key);
if (token != null) {
value[0] = token;
return true;
}
return false;
}
public void setValue(String key, SessionToken value) {
if (key != null) {
cache.put(key, value);
}
}
public void deleteByPrefix(String prefix) {
cache.keySet().removeIf(key -> key.startsWith(prefix));
}
}

15
client/src/main/java/info/frostfs/sdk/pool/Statistic.java Normal file
View file

@ -0,0 +1,15 @@
package info.frostfs.sdk.pool;
import lombok.Getter;
import lombok.Setter;
import java.util.ArrayList;
import java.util.List;
@Getter
@Setter
public class Statistic {
private long overallErrors;
private List<NodeStatistic> nodes = new ArrayList<>();
private String[] currentNodes;
}

13
client/src/main/java/info/frostfs/sdk/pool/StatusSnapshot.java Normal file
View file

@ -0,0 +1,13 @@
package info.frostfs.sdk.pool;
import lombok.Getter;
import lombok.Setter;
@Getter
@Setter
public class StatusSnapshot {
private long allTime;
private long allRequests;
}

22
client/src/main/java/info/frostfs/sdk/pool/WorkList.java Normal file
View file

@ -0,0 +1,22 @@
package info.frostfs.sdk.pool;
import java.util.ArrayList;
import java.util.List;
class WorkList {
private final List<Integer> elements = new ArrayList<>();
private int getLength() {
return elements.size();
}
private void add(int element) {
elements.add(element);
}
private int remove() {
int last = elements.get(elements.size() - 1);
elements.remove(elements.size() - 1);
return last;
}
}

26
client/src/main/java/info/frostfs/sdk/pool/WrapperPrm.java Normal file
View file

@ -0,0 +1,26 @@
package info.frostfs.sdk.pool;
import info.frostfs.sdk.jdo.ECDsa;
import io.grpc.ClientInterceptors;
import io.grpc.ManagedChannelBuilder;
import lombok.Getter;
import lombok.Setter;
import org.slf4j.Logger;
import java.util.Collection;
@Getter
@Setter
public class WrapperPrm {
private Logger logger;
private String address;
private ECDsa key;
private long dialTimeout;
private long streamTimeout;
private int errorThreshold;
private Runnable responseInfoCallback;
private Runnable poolRequestInfoCallback;
private ManagedChannelBuilder<?> grpcChannelOptions;
private long gracefulCloseOnSwitchTimeout;
private Collection<ClientInterceptors> interceptors;
}

7
client/src/main/java/info/frostfs/sdk/services/AccountingClient.java Normal file
View file

@ -0,0 +1,7 @@
package info.frostfs.sdk.services;
import frostfs.accounting.Types;
public interface AccountingClient {
Types.Decimal getBalance();
}

5
client/src/main/java/info/frostfs/sdk/services/CommonClient.java Normal file
View file

@ -0,0 +1,5 @@
package info.frostfs.sdk.services;
public interface CommonClient extends
AccountingClient, ApeManagerClient, ContainerClient, NetmapClient, ObjectClient, SessionClient, ToolsClient {
}

46
client/src/main/java/info/frostfs/sdk/services/impl/AccountingClientImpl.java Normal file
View file

@ -0,0 +1,46 @@
package info.frostfs.sdk.services.impl;
import frostfs.accounting.AccountingServiceGrpc;
import frostfs.accounting.Service;
import frostfs.accounting.Types;
import info.frostfs.sdk.jdo.ClientEnvironment;
import info.frostfs.sdk.mappers.object.OwnerIdMapper;
import info.frostfs.sdk.services.AccountingClient;
import info.frostfs.sdk.services.ContextAccessor;
import info.frostfs.sdk.tools.RequestConstructor;
import info.frostfs.sdk.tools.RequestSigner;
import info.frostfs.sdk.tools.Verifier;
import java.util.Map;
public class AccountingClientImpl extends ContextAccessor implements AccountingClient {
private final AccountingServiceGrpc.AccountingServiceBlockingStub serviceBlockingStub;
public AccountingClientImpl(ClientEnvironment clientEnvironment) {
super(clientEnvironment);
this.serviceBlockingStub = AccountingServiceGrpc.newBlockingStub(getContext().getChannel());
}
@Override
public Types.Decimal getBalance() {
var request = createGetRequest(null);
var response = serviceBlockingStub.balance(request);
Verifier.checkResponse(response);
return response.getBody().getBalance();
}
private Service.BalanceRequest createGetRequest(Map<String, String> xHeaders) {
var body = Service.BalanceRequest.Body.newBuilder()
.setOwnerId(OwnerIdMapper.toGrpcMessage(getContext().getOwnerId()))
.build();
var request = Service.BalanceRequest.newBuilder()
.setBody(body);
RequestConstructor.addMetaHeader(request, xHeaders);
RequestSigner.sign(request, getContext().getKey());
return request.build();
}
}

16
client/src/main/java/info/frostfs/sdk/services/impl/ObjectToolsImpl.java
View file

@ -47,16 +47,16 @@ public class ObjectToolsImpl extends ContextAccessor implements ToolsClient {
);
}
public Types.Object createObject(ObjectFrostFS objectFrostFs) {
objectFrostFs.getHeader().setOwnerId(getContext().getOwnerId());
objectFrostFs.getHeader().setVersion(getContext().getVersion());
objectFrostFs.getHeader().setPayloadLength(objectFrostFs.getPayload().length);
public Types.Object createObject(ObjectFrostFS objectFrostFS) {
objectFrostFS.getHeader().setOwnerId(getContext().getOwnerId());
objectFrostFS.getHeader().setVersion(getContext().getVersion());
objectFrostFS.getHeader().setPayloadLength(objectFrostFS.getPayload().length);
var grpcHeader = ObjectHeaderMapper.toGrpcMessage(objectFrostFs.getHeader()).toBuilder()
.setPayloadHash(sha256Checksum(objectFrostFs.getPayload()))
var grpcHeader = ObjectHeaderMapper.toGrpcMessage(objectFrostFS.getHeader()).toBuilder()
.setPayloadHash(sha256Checksum(objectFrostFS.getPayload()))
.build();
var split = objectFrostFs.getHeader().getSplit();
var split = objectFrostFS.getHeader().getSplit();
if (nonNull(split)) {
grpcHeader = updateSplitValues(grpcHeader, split);
}
@ -68,7 +68,7 @@ public class ObjectToolsImpl extends ContextAccessor implements ToolsClient {
return Types.Object.newBuilder()
.setHeader(grpcHeader)
.setObjectId(objectId)
.setPayload(ByteString.copyFrom(objectFrostFs.getPayload()))
.setPayload(ByteString.copyFrom(objectFrostFS.getPayload()))
.setSignature(sig)
.build();
}

12
client/src/main/java/info/frostfs/sdk/services/impl/interceptor/ClientMetrics.java
View file

@ -15,10 +15,10 @@ import static info.frostfs.sdk.services.impl.interceptor.Labels.*;
public class ClientMetrics {
private static final List<String> defaultRequestLabels =
Arrays.asList("grpc_type", "grpc_service", "grpc_method");
Arrays.asList("grpc_target", "grpc_type", "grpc_service", "grpc_method");
private static final List<String> defaultResponseLabels =
Arrays.asList("grpc_type", "grpc_service", "grpc_method", "code", "grpc_code");
Arrays.asList("grpc_target", "grpc_type", "grpc_service", "grpc_method", "code", "grpc_code");
private static final Counter.Builder rpcStartedBuilder =
Counter.build()
@ -113,7 +113,9 @@ public class ClientMetrics {
.observe(latencySec);
}
/** Knows how to produce {@link ClientMetrics} instances for individual methods. */
/**
* Knows how to produce {@link ClientMetrics} instances for individual methods.
*/
static class Factory {
private final List<Metadata.Key<String>> labelHeaderKeys;
private final Counter rpcStarted;
@ -155,7 +157,9 @@ public class ClientMetrics {
}
}
/** Creates a {@link ClientMetrics} for the supplied gRPC method. */
/**
* Creates a {@link ClientMetrics} for the supplied gRPC method.
*/
ClientMetrics createMetricsForMethod(GrpcMethod grpcMethod) {
return new ClientMetrics(
labelHeaderKeys,

12
client/src/main/java/info/frostfs/sdk/services/impl/interceptor/GrpcMethod.java
View file

@ -1,27 +1,33 @@
package info.frostfs.sdk.services.impl.interceptor;
import io.grpc.Channel;
import io.grpc.MethodDescriptor;
import io.grpc.MethodDescriptor.MethodType;
public class GrpcMethod {
private final String targetName;
private final String serviceName;
private final String methodName;
private final MethodType type;
private GrpcMethod(String serviceName, String methodName, MethodType type) {
private GrpcMethod(String targetName, String serviceName, String methodName, MethodType type) {
this.targetName = targetName;
this.serviceName = serviceName;
this.methodName = methodName;
this.type = type;
}
static GrpcMethod of(MethodDescriptor<?, ?> method) {
static GrpcMethod of(MethodDescriptor<?, ?> method, Channel channel) {
String serviceName = MethodDescriptor.extractFullServiceName(method.getFullMethodName());
// Full method names are of the form: "full.serviceName/MethodName". We extract the last part.
String methodName = method.getFullMethodName().substring(serviceName.length() + 1);
return new GrpcMethod(serviceName, methodName, method.getType());
return new GrpcMethod(channel.authority(), serviceName, methodName, method.getType());
}
String targetName() {
return targetName;
}
String serviceName() {
return serviceName;

1
client/src/main/java/info/frostfs/sdk/services/impl/interceptor/Labels.java
View file

@ -53,6 +53,7 @@ public class Labels {
*/
static <T> T addLabels(SimpleCollector<T> collector, List<String> labels, GrpcMethod method) {
List<String> allLabels = new ArrayList<>();
allLabels.add(method.targetName());
allLabels.add(method.type());
allLabels.add(method.serviceName());
allLabels.add(method.methodName());

10
client/src/main/java/info/frostfs/sdk/services/impl/interceptor/MonitoringClientInterceptor.java
View file

@ -1,11 +1,8 @@
package info.frostfs.sdk.services.impl.interceptor;
import io.grpc.*;
import java.time.Clock;
import io.grpc.CallOptions;
import io.grpc.Channel;
import io.grpc.ClientCall;
import io.grpc.ClientInterceptor;
import io.grpc.MethodDescriptor;
public class MonitoringClientInterceptor implements ClientInterceptor {
@ -19,6 +16,7 @@ public class MonitoringClientInterceptor implements ClientInterceptor {
this.configuration = configuration;
this.clientMetricsFactory = clientMetricsFactory;
}
public static MonitoringClientInterceptor create(Configuration configuration) {
return new MonitoringClientInterceptor(
Clock.systemDefaultZone(), configuration, new ClientMetrics.Factory(configuration));
@ -27,7 +25,7 @@ public class MonitoringClientInterceptor implements ClientInterceptor {
@Override
public <R, S> ClientCall<R, S> interceptCall(
MethodDescriptor<R, S> methodDescriptor, CallOptions callOptions, Channel channel) {
GrpcMethod grpcMethod = GrpcMethod.of(methodDescriptor);
GrpcMethod grpcMethod = GrpcMethod.of(methodDescriptor, channel);
ClientMetrics metrics = clientMetricsFactory.createMetricsForMethod(grpcMethod);
return new MonitoringClientCall<>(
channel.newCall(methodDescriptor, callOptions), metrics, grpcMethod, configuration, clock);

15
client/src/main/java/info/frostfs/sdk/tools/GrpcClient.java
View file

@ -3,7 +3,7 @@ package info.frostfs.sdk.tools;
import info.frostfs.sdk.exceptions.ValidationFrostFSException;
import info.frostfs.sdk.jdo.ClientSettings;
import info.frostfs.sdk.utils.Validator;
import io.grpc.Channel;
import io.grpc.ManagedChannel;
import io.grpc.netty.NettyChannelBuilder;
import java.net.URI;
@ -16,7 +16,7 @@ public class GrpcClient {
private GrpcClient() {
}
public static Channel initGrpcChannel(ClientSettings clientSettings) {
public static ManagedChannel initGrpcChannel(ClientSettings clientSettings) {
Validator.validate(clientSettings);
try {
@ -32,4 +32,15 @@ public class GrpcClient {
);
}
}
public static ManagedChannel initGrpcChannel(String address) {
try {
URI uri = new URI(address);
return NettyChannelBuilder.forAddress(uri.getHost(), uri.getPort()).usePlaintext().build();
} catch (URISyntaxException exp) {
throw new ValidationFrostFSException(
String.format(INVALID_HOST_TEMPLATE, address, exp.getMessage())
);
}
}
}

20
client/src/main/java/info/frostfs/sdk/utils/FrostFSMessages.java Normal file
View file

@ -0,0 +1,20 @@
package info.frostfs.sdk.utils;
import org.slf4j.Logger;
public class FrostFSMessages {
private FrostFSMessages() {
}
public static void sessionCreationError(Logger logger, String address, String error) {
logger.warn("Failed to create frostfs session token for client. Address {}, {}", address, error);
}
public static void errorThresholdReached(Logger logger, String address, long threshold) {
logger.warn("Error threshold reached. Address {}, threshold {}", address, threshold);
}
public static void healthChanged(Logger logger, String address, boolean healthy, String error) {
logger.warn("Health has changed: {} healthy {}, reason {}", address, healthy, error);
}
}

102
client/src/test/java/info/frostfs/sdk/services/AccountingClientTest.java Normal file
View file

@ -0,0 +1,102 @@
package info.frostfs.sdk.services;
import frostfs.accounting.AccountingServiceGrpc;
import frostfs.accounting.Service;
import info.frostfs.sdk.Base58;
import info.frostfs.sdk.dto.object.OwnerId;
import info.frostfs.sdk.jdo.ClientEnvironment;
import info.frostfs.sdk.services.impl.AccountingClientImpl;
import info.frostfs.sdk.testgenerator.AccountingGenerator;
import info.frostfs.sdk.tools.RequestConstructor;
import info.frostfs.sdk.tools.RequestSigner;
import info.frostfs.sdk.tools.Verifier;
import io.grpc.Channel;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.junit.jupiter.api.extension.ExtendWith;
import org.junit.platform.commons.util.ReflectionUtils;
import org.mockito.ArgumentCaptor;
import org.mockito.Mock;
import org.mockito.MockedStatic;
import org.mockito.Mockito;
import org.mockito.junit.jupiter.MockitoExtension;
import java.lang.reflect.Field;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.mockito.Mockito.*;
@ExtendWith(MockitoExtension.class)
class AccountingClientTest {
private static final String OWNER_ID = "NVxUSpEEJzYXZZtUs18PrJTD9QZkLLNQ8S";
private AccountingClientImpl accountingClient;
@Mock
private AccountingServiceGrpc.AccountingServiceBlockingStub AccountingServiceClient;
@Mock
private ClientEnvironment clientEnvironment;
@Mock
private Channel channel;
private MockedStatic<Verifier> verifierMock;
private MockedStatic<RequestConstructor> requestConstructorMock;
private MockedStatic<RequestSigner> requestSignerMock;
@BeforeEach
void setUp() throws IllegalAccessException {
when(clientEnvironment.getChannel()).thenReturn(channel);
when(clientEnvironment.getOwnerId()).thenReturn(new OwnerId(OWNER_ID));
accountingClient = new AccountingClientImpl(clientEnvironment);
Field field = ReflectionUtils.findFields(AccountingClientImpl.class,
f -> f.getName().equals("serviceBlockingStub"),
ReflectionUtils.HierarchyTraversalMode.TOP_DOWN)
.get(0);
field.setAccessible(true);
field.set(accountingClient, AccountingServiceClient);
verifierMock = Mockito.mockStatic(Verifier.class);
requestConstructorMock = Mockito.mockStatic(RequestConstructor.class);
requestSignerMock = Mockito.mockStatic(RequestSigner.class);
}
@AfterEach
void cleanUp() {
verifierMock.close();
requestConstructorMock.close();
requestSignerMock.close();
}
@Test
void getBalance_success() {
//Given
var response = AccountingGenerator.generateBalanceResponse();
var captor = ArgumentCaptor.forClass(Service.BalanceRequest.class);
when(AccountingServiceClient.balance(captor.capture())).thenReturn(response);
//When
var result = accountingClient.getBalance();
//Then
requestConstructorMock.verify(
() -> RequestConstructor.addMetaHeader(any(Service.BalanceRequest.Builder.class), eq(null)),
times(1)
);
requestSignerMock.verify(
() -> RequestSigner.sign(any(Service.BalanceRequest.Builder.class), eq(null)),
times(1)
);
verifierMock.verify(() -> Verifier.checkResponse(response), times(1));
assertEquals(response.getBody().getBalance(), result);
var request = captor.getValue();
assertEquals(OWNER_ID, Base58.encode(request.getBody().getOwnerId().getValue().toByteArray()));
}
}

42
client/src/test/java/info/frostfs/sdk/testgenerator/AccountingGenerator.java Normal file
View file

@ -0,0 +1,42 @@
package info.frostfs.sdk.testgenerator;
import frostfs.accounting.Service;
import frostfs.accounting.Types;
public class AccountingGenerator {
public static Service.BalanceRequest generateBalanceRequest() {
return Service.BalanceRequest.newBuilder()
.setBody(generateBalanceRequestBody())
.setMetaHeader(SessionGenerator.generateRequestMetaHeader())
.setVerifyHeader(SessionGenerator.generateRequestVerificationHeader())
.build();
}
public static Service.BalanceRequest.Body generateBalanceRequestBody() {
return Service.BalanceRequest.Body.newBuilder()
.setOwnerId(RefsGenerator.generateOwnerID())
.build();
}
public static Service.BalanceResponse generateBalanceResponse() {
return Service.BalanceResponse.newBuilder()
.setBody(generateBalanceResponseBody())
.setMetaHeader(SessionGenerator.generateResponseMetaHeader())
.setVerifyHeader(SessionGenerator.generateResponseVerificationHeader())
.build();
}
public static Service.BalanceResponse.Body generateBalanceResponseBody() {
return Service.BalanceResponse.Body.newBuilder()
.setBalance(generateDecimal())
.build();
}
public static Types.Decimal generateDecimal() {
return Types.Decimal.newBuilder()
.setValue(1)
.setPrecision(2)
.build();
}
}

4
cryptography/pom.xml
View file

@ -6,7 +6,7 @@
<parent>
<groupId>info.frostfs.sdk</groupId>
<artifactId>frostfs-sdk-java</artifactId>
<version>0.1.0</version>
<version>0.2.0</version>
</parent>
<artifactId>cryptography</artifactId>
@ -21,7 +21,7 @@
<dependency>
<groupId>info.frostfs.sdk</groupId>
<artifactId>exceptions</artifactId>
<version>0.1.0</version>
<version>0.2.0</version>
</dependency>
<dependency>
<groupId>com.google.protobuf</groupId>

2
exceptions/pom.xml
View file

@ -6,7 +6,7 @@
<parent>
<groupId>info.frostfs.sdk</groupId>
<artifactId>frostfs-sdk-java</artifactId>
<version>0.1.0</version>
<version>0.2.0</version>
</parent>
<artifactId>exceptions</artifactId>

5
exceptions/src/main/java/info/frostfs/sdk/constants/ErrorConst.java
View file

@ -40,6 +40,11 @@ public class ErrorConst {
public static final String WRONG_UUID_SIZE_TEMPLATE = "uuid byte array length must be %s";
public static final String POOL_CLIENT_UNHEALTHY = "pool client unhealthy";
public static final String POOL_PEERS_IS_MISSING = "no FrostFS peers configured";
public static final String POOL_NODES_UNHEALTHY = "at least one node must be healthy";
public static final String POOL_CLIENTS_UNHEALTHY = "cannot find alive client";
public static final String POOL_NOT_DIALED = "pool not dialed";
public static final String FIELDS_DELIMITER_COMMA = ", ";
public static final String FIELDS_DELIMITER_OR = " or ";

15
exceptions/src/main/java/info/frostfs/sdk/exceptions/FrostFSException.java Normal file
View file

@ -0,0 +1,15 @@
package info.frostfs.sdk.exceptions;
public class FrostFSException extends RuntimeException {
public FrostFSException() {
}
public FrostFSException(String message) {
super(message);
}
public FrostFSException(Throwable cause) {
super(cause);
}
}

2
exceptions/src/main/java/info/frostfs/sdk/exceptions/ProcessFrostFSException.java
View file

@ -1,6 +1,6 @@
package info.frostfs.sdk.exceptions;
public class ProcessFrostFSException extends RuntimeException {
public class ProcessFrostFSException extends FrostFSException {
public ProcessFrostFSException(String message) {
super(message);
}

8
exceptions/src/main/java/info/frostfs/sdk/exceptions/SessionExpiredFrostFSException.java Normal file
View file

@ -0,0 +1,8 @@
package info.frostfs.sdk.exceptions;
public class SessionExpiredFrostFSException extends FrostFSException {
public SessionExpiredFrostFSException(String message) {
super(message);
}
}

8
exceptions/src/main/java/info/frostfs/sdk/exceptions/SessionNotFoundFrostFSException.java Normal file
View file

@ -0,0 +1,8 @@
package info.frostfs.sdk.exceptions;
public class SessionNotFoundFrostFSException extends FrostFSException {
public SessionNotFoundFrostFSException(String message) {
super(message);
}
}

2
exceptions/src/main/java/info/frostfs/sdk/exceptions/TimeoutFrostFSException.java
View file

@ -1,6 +1,6 @@
package info.frostfs.sdk.exceptions;
public class TimeoutFrostFSException extends RuntimeException {
public class TimeoutFrostFSException extends FrostFSException {
public TimeoutFrostFSException() {
}
}

2
exceptions/src/main/java/info/frostfs/sdk/exceptions/ValidationFrostFSException.java
View file

@ -1,6 +1,6 @@
package info.frostfs.sdk.exceptions;
public class ValidationFrostFSException extends RuntimeException {
public class ValidationFrostFSException extends FrostFSException {
public ValidationFrostFSException(String message) {
super(message);
}

8
models/pom.xml
View file

@ -6,7 +6,7 @@
<parent>
<groupId>info.frostfs.sdk</groupId>
<artifactId>frostfs-sdk-java</artifactId>
<version>0.1.0</version>
<version>0.2.0</version>
</parent>
<artifactId>models</artifactId>
@ -21,17 +21,17 @@
<dependency>
<groupId>info.frostfs.sdk</groupId>
<artifactId>cryptography</artifactId>
<version>0.1.0</version>
<version>0.2.0</version>
</dependency>
<dependency>
<groupId>info.frostfs.sdk</groupId>
<artifactId>protos</artifactId>
<version>0.1.0</version>
<version>0.2.0</version>
</dependency>
<dependency>
<groupId>info.frostfs.sdk</groupId>
<artifactId>exceptions</artifactId>
<version>0.1.0</version>
<version>0.2.0</version>
</dependency>
</dependencies>

2
models/src/main/java/info/frostfs/sdk/constants/AppConst.java
View file

@ -1,6 +1,8 @@
package info.frostfs.sdk.constants;
public class AppConst {
public static final String RESERVED_PREFIX = "__SYSTEM__";
public static final int DEFAULT_MAJOR_VERSION = 2;
public static final int DEFAULT_MINOR_VERSION = 13;
public static final int BYTE_SHIFT = 10;

10
models/src/main/java/info/frostfs/sdk/constants/AttributeConst.java Normal file
View file

@ -0,0 +1,10 @@
package info.frostfs.sdk.constants;
import static info.frostfs.sdk.constants.AppConst.RESERVED_PREFIX;
public class AttributeConst {
public static final String DISABLE_HOMOMORPHIC_HASHING_ATTRIBUTE = RESERVED_PREFIX + "DISABLE_HOMOMORPHIC_HASHING";
private AttributeConst() {
}
}

7
models/src/main/java/info/frostfs/sdk/constants/XHeaderConst.java
View file

@ -1,9 +1,10 @@
package info.frostfs.sdk.constants;
import static info.frostfs.sdk.constants.AppConst.RESERVED_PREFIX;
public class XHeaderConst {
public static final String RESERVED_XHEADER_PREFIX = "__SYSTEM__";
public static final String XHEADER_NETMAP_EPOCH = RESERVED_XHEADER_PREFIX + "NETMAP_EPOCH";
public static final String XHEADER_NETMAP_LOOKUP_DEPTH = RESERVED_XHEADER_PREFIX + "NETMAP_LOOKUP_DEPTH";
public static final String XHEADER_NETMAP_EPOCH = RESERVED_PREFIX + "NETMAP_EPOCH";
public static final String XHEADER_NETMAP_LOOKUP_DEPTH = RESERVED_PREFIX + "NETMAP_LOOKUP_DEPTH";
private XHeaderConst() {
}

3
models/src/main/java/info/frostfs/sdk/dto/container/Container.java
View file

@ -6,6 +6,8 @@ import info.frostfs.sdk.enums.BasicAcl;
import lombok.Getter;
import lombok.Setter;
import java.util.HashMap;
import java.util.Map;
import java.util.UUID;
@Getter
@ -15,6 +17,7 @@ public class Container {
private BasicAcl basicAcl;
private PlacementPolicy placementPolicy;
private Version version;
private Map<String, String> attributes = new HashMap<>();
public Container(BasicAcl basicAcl, PlacementPolicy placementPolicy) {
this.nonce = UUID.randomUUID();

30
models/src/main/java/info/frostfs/sdk/mappers/container/ContainerMapper.java
View file

@ -7,9 +7,13 @@ import info.frostfs.sdk.enums.BasicAcl;
import info.frostfs.sdk.exceptions.ProcessFrostFSException;
import info.frostfs.sdk.mappers.netmap.PlacementPolicyMapper;
import info.frostfs.sdk.mappers.netmap.VersionMapper;
import org.apache.commons.collections4.CollectionUtils;
import java.util.stream.Collectors;
import static info.frostfs.sdk.UuidExtension.asBytes;
import static info.frostfs.sdk.UuidExtension.asUuid;
import static info.frostfs.sdk.constants.AttributeConst.DISABLE_HOMOMORPHIC_HASHING_ATTRIBUTE;
import static info.frostfs.sdk.constants.ErrorConst.UNKNOWN_ENUM_VALUE_TEMPLATE;
import static java.util.Objects.isNull;
@ -22,11 +26,23 @@ public class ContainerMapper {
return null;
}
return Types.Container.newBuilder()
var containerGrpc = Types.Container.newBuilder()
.setBasicAcl(container.getBasicAcl().value)
.setPlacementPolicy(PlacementPolicyMapper.toGrpcMessage(container.getPlacementPolicy()))
.setNonce(ByteString.copyFrom(asBytes(container.getNonce())))
.build();
.setNonce(ByteString.copyFrom(asBytes(container.getNonce())));
container.getAttributes().putIfAbsent(DISABLE_HOMOMORPHIC_HASHING_ATTRIBUTE, Boolean.TRUE.toString());
var attributes = container.getAttributes().entrySet().stream()
.map(entry ->
Types.Container.Attribute.newBuilder()
.setKey(entry.getKey())
.setValue(entry.getValue())
.build()
)
.collect(Collectors.toList());
containerGrpc.addAllAttributes(attributes);
return containerGrpc.build();
}
public static Container toModel(Types.Container containerGrpc) {
@ -41,9 +57,17 @@ public class ContainerMapper {
);
}
var container = new Container(basicAcl, PlacementPolicyMapper.toModel(containerGrpc.getPlacementPolicy()));
container.setNonce(asUuid(containerGrpc.getNonce().toByteArray()));
container.setVersion(VersionMapper.toModel(containerGrpc.getVersion()));
if (CollectionUtils.isNotEmpty(containerGrpc.getAttributesList())) {
var attributes = containerGrpc.getAttributesList().stream()
.collect(Collectors.toMap(Types.Container.Attribute::getKey, Types.Container.Attribute::getValue));
container.setAttributes(attributes);
}
return container;
}
}

23
models/src/test/java/info/frostfs/sdk/mappers/container/ContainerMapperTest.java
View file

@ -15,6 +15,7 @@ import java.util.UUID;
import static info.frostfs.sdk.UuidExtension.asBytes;
import static info.frostfs.sdk.UuidExtension.asUuid;
import static info.frostfs.sdk.constants.AttributeConst.DISABLE_HOMOMORPHIC_HASHING_ATTRIBUTE;
import static org.junit.jupiter.api.Assertions.*;
public class ContainerMapperTest {
@ -24,6 +25,8 @@ public class ContainerMapperTest {
//Given
var placementPolicy = new PlacementPolicy(new Replica[]{new Replica(1)}, true);
var container = new Container(BasicAcl.PUBLIC_RW, placementPolicy);
container.getAttributes().put("key1", "val1");
container.getAttributes().put(DISABLE_HOMOMORPHIC_HASHING_ATTRIBUTE, "false");
//When
var result = ContainerMapper.toGrpcMessage(container);
@ -42,6 +45,11 @@ public class ContainerMapperTest {
container.getPlacementPolicy().getReplicas()[0].getSelector(),
result.getPlacementPolicy().getReplicasList().get(0).getSelector()
);
assertEquals("key1", result.getAttributes(0).getKey());
assertEquals("val1", result.getAttributes(0).getValue());
assertEquals(DISABLE_HOMOMORPHIC_HASHING_ATTRIBUTE, result.getAttributes(1).getKey());
assertEquals("false", result.getAttributes(1).getValue());
}
@Test
@ -69,11 +77,23 @@ public class ContainerMapperTest {
.addReplicas(replica)
.build();
var attribute1 = Types.Container.Attribute.newBuilder()
.setKey("key1")
.setValue("val1")
.build();
var attribute2 = Types.Container.Attribute.newBuilder()
.setKey("key2")
.setValue("val2")
.build();
var container = Types.Container.newBuilder()
.setBasicAcl(basicAcl.value)
.setNonce(ByteString.copyFrom(asBytes(UUID.randomUUID())))
.setVersion(version)
.setPlacementPolicy(placementPolicy)
.addAttributes(attribute1)
.addAttributes(attribute2)
.build();
//When
@ -95,6 +115,9 @@ public class ContainerMapperTest {
);
assertEquals(version.getMajor(), result.getVersion().getMajor());
assertEquals(version.getMinor(), result.getVersion().getMinor());
assertEquals(attribute1.getValue(), result.getAttributes().get(attribute1.getKey()));
assertEquals(attribute2.getValue(), result.getAttributes().get(attribute2.getKey()));
}
@Test

2
pom.xml
View file

@ -6,7 +6,7 @@
<groupId>info.frostfs.sdk</groupId>
<artifactId>frostfs-sdk-java</artifactId>
<version>0.1.0</version>
<version>0.2.0</version>
<packaging>pom</packaging>
<modules>
<module>client</module>

2
protos/pom.xml
View file

@ -6,7 +6,7 @@
<parent>
<groupId>info.frostfs.sdk</groupId>
<artifactId>frostfs-sdk-java</artifactId>
<version>0.1.0</version>
<version>0.2.0</version>
</parent>
<artifactId>protos</artifactId>