diff --git a/run-wrapper/update b/run-wrapper/update index 77b3137b..63c1df21 100755 --- a/run-wrapper/update +++ b/run-wrapper/update @@ -23,8 +23,8 @@ echo Downloading... cd "$SCRIPT_DIR" -rm "Run wrapper.zip" -rm -rf "dhfs" +rm "Run wrapper.zip" || true +rm -rf "dhfs" || true wget https://nightly.link/usatiuk/dhfs/actions/runs/$LATEST/Run%20wrapper.zip diff --git a/server/pom.xml b/server/pom.xml index 664ba8ee..bda73dfe 100644 --- a/server/pom.xml +++ b/server/pom.xml @@ -73,6 +73,14 @@ io.quarkus quarkus-rest + + io.quarkus + quarkus-rest-client + + + io.quarkus + quarkus-rest-client-jsonb + io.quarkus quarkus-rest-jsonb diff --git a/server/src/main/java/com/usatiuk/dhfs/objects/repository/RemoteHostManager.java b/server/src/main/java/com/usatiuk/dhfs/objects/repository/RemoteHostManager.java index 2a491a72..47611d03 100644 --- a/server/src/main/java/com/usatiuk/dhfs/objects/repository/RemoteHostManager.java +++ b/server/src/main/java/com/usatiuk/dhfs/objects/repository/RemoteHostManager.java @@ -1,6 +1,6 @@ package com.usatiuk.dhfs.objects.repository; -import com.usatiuk.dhfs.objects.repository.peersync.GetSelfInfoRequest; +import com.usatiuk.dhfs.objects.repository.peersync.PeerSyncApiClientDynamic; import com.usatiuk.dhfs.objects.repository.peersync.PersistentPeerInfo; import com.usatiuk.dhfs.objects.repository.webapi.AvailablePeerInfo; import io.quarkus.logging.Log; @@ -16,10 +16,7 @@ import org.eclipse.microprofile.config.inject.ConfigProperty; import java.io.IOException; import java.security.cert.CertificateException; -import java.util.Collection; -import java.util.List; -import java.util.Map; -import java.util.UUID; +import java.util.*; import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentMap; @@ -33,6 +30,8 @@ public class RemoteHostManager { SyncHandler syncHandler; @Inject RpcClientFactory rpcClientFactory; + @Inject + PeerSyncApiClientDynamic peerSyncApiClient; @ConfigProperty(name = "dhfs.objects.sync.ping.timeout") long pingTimeout; boolean _initialized = false; @@ -184,13 +183,12 @@ public class RemoteHostManager { // FIXME: race? - var info = rpcClientFactory.withPeerSyncClient(state.getAddr(), state.getPort(), 10000L, c -> { - return c.getSelfInfo(GetSelfInfoRequest.getDefaultInstance()); - }); + var info = peerSyncApiClient.getSelfInfo(state.getAddr(), state.getPort()); try { persistentRemoteHostsService.addHost( - new PersistentPeerInfo(UUID.fromString(info.getUuid()), CertificateTools.certFromBytes(info.getCert().toByteArray()))); + new PersistentPeerInfo(UUID.fromString(info.selfUuid()), + CertificateTools.certFromBytes(Base64.getDecoder().decode(info.cert())))); Log.info("Added host: " + host.toString()); } catch (CertificateException e) { throw new RuntimeException(e); diff --git a/server/src/main/java/com/usatiuk/dhfs/objects/repository/RpcClientFactory.java b/server/src/main/java/com/usatiuk/dhfs/objects/repository/RpcClientFactory.java index 884aa5c5..b69c548b 100644 --- a/server/src/main/java/com/usatiuk/dhfs/objects/repository/RpcClientFactory.java +++ b/server/src/main/java/com/usatiuk/dhfs/objects/repository/RpcClientFactory.java @@ -1,6 +1,5 @@ package com.usatiuk.dhfs.objects.repository; -import com.usatiuk.dhfs.objects.repository.peersync.DhfsObjectPeerSyncGrpcGrpc; import io.grpc.Status; import io.grpc.StatusRuntimeException; import io.quarkus.logging.Log; @@ -32,7 +31,6 @@ public class RpcClientFactory { RpcChannelFactory rpcChannelFactory; // FIXME: Leaks! private ConcurrentMap _objSyncCache = new ConcurrentHashMap<>(); - private ConcurrentMap _peerSyncCache = new ConcurrentHashMap<>(); public R withObjSyncClient(Collection targets, ObjectSyncClientFunction fn) { var shuffledList = new ArrayList<>(targets); @@ -85,27 +83,9 @@ public class RpcClientFactory { return fn.apply(stub.withDeadlineAfter(timeout, TimeUnit.SECONDS)); } - public R withPeerSyncClient(UUID target, PeerSyncClientFunction fn) { - var hostinfo = remoteHostManager.getTransientState(target); - if (hostinfo.getAddr() == null) throw new IllegalStateException("Address for " + target + " not yet known"); - return withPeerSyncClient(hostinfo.getAddr(), hostinfo.getPort(), peerSyncTimeout, fn); - } - - public R withPeerSyncClient(String addr, int port, long timeout, PeerSyncClientFunction fn) { - var key = new PeerSyncStubKey(addr, port); - var stub = _peerSyncCache.computeIfAbsent(key, (k) -> { - var channel = rpcChannelFactory.getInsecureChannel(addr, port); - return DhfsObjectPeerSyncGrpcGrpc.newBlockingStub(channel) - .withMaxOutboundMessageSize(Integer.MAX_VALUE) - .withMaxInboundMessageSize(Integer.MAX_VALUE); - }); - return fn.apply(stub.withDeadlineAfter(timeout, TimeUnit.SECONDS)); - } - public void dropCache() { rpcChannelFactory.dropCache(); _objSyncCache = new ConcurrentHashMap<>(); - _peerSyncCache = new ConcurrentHashMap<>(); } @FunctionalInterface @@ -113,15 +93,7 @@ public class RpcClientFactory { R apply(DhfsObjectSyncGrpcGrpc.DhfsObjectSyncGrpcBlockingStub client); } - @FunctionalInterface - public interface PeerSyncClientFunction { - R apply(DhfsObjectPeerSyncGrpcGrpc.DhfsObjectPeerSyncGrpcBlockingStub client); - } - private record ObjSyncStubKey(String host, String address, int port) { } - private record PeerSyncStubKey(String address, int port) { - } - } diff --git a/server/src/main/java/com/usatiuk/dhfs/objects/repository/peersync/PeerInfo.java b/server/src/main/java/com/usatiuk/dhfs/objects/repository/peersync/PeerInfo.java new file mode 100644 index 00000000..e51e4d02 --- /dev/null +++ b/server/src/main/java/com/usatiuk/dhfs/objects/repository/peersync/PeerInfo.java @@ -0,0 +1,4 @@ +package com.usatiuk.dhfs.objects.repository.peersync; + +public record PeerInfo(String selfUuid, String cert) { +} diff --git a/server/src/main/java/com/usatiuk/dhfs/objects/repository/peersync/PeerSyncApi.java b/server/src/main/java/com/usatiuk/dhfs/objects/repository/peersync/PeerSyncApi.java new file mode 100644 index 00000000..0fadf3e3 --- /dev/null +++ b/server/src/main/java/com/usatiuk/dhfs/objects/repository/peersync/PeerSyncApi.java @@ -0,0 +1,26 @@ +package com.usatiuk.dhfs.objects.repository.peersync; + +import com.usatiuk.dhfs.objects.repository.PersistentRemoteHostsService; +import jakarta.inject.Inject; +import jakarta.ws.rs.GET; +import jakarta.ws.rs.Path; + +import java.security.cert.CertificateEncodingException; +import java.util.Base64; + +@Path("/peer-info") +public class PeerSyncApi { + @Inject + PersistentRemoteHostsService persistentRemoteHostsService; + + @Path("self") + @GET + public PeerInfo getSelfInfo() { + try { + return new PeerInfo(persistentRemoteHostsService.getSelfUuid().toString(), + Base64.getEncoder().encodeToString(persistentRemoteHostsService.getSelfCertificate().getEncoded())); + } catch (CertificateEncodingException e) { + throw new RuntimeException(e); + } + } +} diff --git a/server/src/main/java/com/usatiuk/dhfs/objects/repository/peersync/PeerSyncApiClient.java b/server/src/main/java/com/usatiuk/dhfs/objects/repository/peersync/PeerSyncApiClient.java new file mode 100644 index 00000000..bb96a480 --- /dev/null +++ b/server/src/main/java/com/usatiuk/dhfs/objects/repository/peersync/PeerSyncApiClient.java @@ -0,0 +1,11 @@ +package com.usatiuk.dhfs.objects.repository.peersync; + +import jakarta.ws.rs.GET; +import jakarta.ws.rs.Path; + +@Path("/peer-info") +public interface PeerSyncApiClient { + @Path("self") + @GET + PeerInfo getSelfInfo(); +} diff --git a/server/src/main/java/com/usatiuk/dhfs/objects/repository/peersync/PeerSyncApiClientDynamic.java b/server/src/main/java/com/usatiuk/dhfs/objects/repository/peersync/PeerSyncApiClientDynamic.java new file mode 100644 index 00000000..f0d39caa --- /dev/null +++ b/server/src/main/java/com/usatiuk/dhfs/objects/repository/peersync/PeerSyncApiClientDynamic.java @@ -0,0 +1,19 @@ +package com.usatiuk.dhfs.objects.repository.peersync; + +import io.quarkus.rest.client.reactive.QuarkusRestClientBuilder; +import jakarta.enterprise.context.ApplicationScoped; + +import java.net.URI; +import java.util.concurrent.TimeUnit; + +@ApplicationScoped +public class PeerSyncApiClientDynamic { + public PeerInfo getSelfInfo(String addr, int port) { + var client = QuarkusRestClientBuilder.newBuilder() + .baseUri(URI.create("http://" + addr + ":" + port)) + .connectTimeout(5, TimeUnit.SECONDS) + .readTimeout(5, TimeUnit.SECONDS) + .build(PeerSyncApiClient.class); + return client.getSelfInfo(); + } +} diff --git a/server/src/main/java/com/usatiuk/dhfs/objects/repository/peersync/PeerSyncServer.java b/server/src/main/java/com/usatiuk/dhfs/objects/repository/peersync/PeerSyncServer.java deleted file mode 100644 index f471299b..00000000 --- a/server/src/main/java/com/usatiuk/dhfs/objects/repository/peersync/PeerSyncServer.java +++ /dev/null @@ -1,30 +0,0 @@ -package com.usatiuk.dhfs.objects.repository.peersync; - -import com.google.protobuf.ByteString; -import com.usatiuk.dhfs.objects.repository.PersistentRemoteHostsService; -import io.quarkus.grpc.GrpcService; -import io.smallrye.common.annotation.Blocking; -import io.smallrye.mutiny.Uni; -import jakarta.inject.Inject; - -import java.security.cert.CertificateEncodingException; - -@GrpcService -public class PeerSyncServer implements DhfsObjectPeerSyncGrpc { - @Inject - PersistentRemoteHostsService persistentRemoteHostsService; - - @Override - @Blocking - public Uni getSelfInfo(GetSelfInfoRequest request) { - try { - return Uni.createFrom().item( - PeerInfo.newBuilder() - .setUuid(persistentRemoteHostsService.getSelfUuid().toString()) - .setCert(ByteString.copyFrom(persistentRemoteHostsService.getSelfCertificate().getEncoded())) - .build()); - } catch (CertificateEncodingException e) { - throw new RuntimeException(e); - } - } -} diff --git a/server/src/main/proto/dhfs_objects_peer_sync.proto b/server/src/main/proto/dhfs_objects_peer_sync.proto deleted file mode 100644 index ebf616e7..00000000 --- a/server/src/main/proto/dhfs_objects_peer_sync.proto +++ /dev/null @@ -1,20 +0,0 @@ -syntax = "proto3"; - -option java_multiple_files = true; -option java_package = "com.usatiuk.dhfs.objects.repository.peersync"; -option java_outer_classname = "DhfsObjectPeerSyncApi"; - -package dhfs.objects.peersync; - -service DhfsObjectPeerSyncGrpc { - rpc GetSelfInfo (GetSelfInfoRequest) returns (PeerInfo) {} -} - -message GetSelfInfoRequest { - -} - -message PeerInfo { - string uuid = 1; - bytes cert = 2; -} \ No newline at end of file diff --git a/server/src/main/resources/application.properties b/server/src/main/resources/application.properties index ebaae794..140ad9a9 100644 --- a/server/src/main/resources/application.properties +++ b/server/src/main/resources/application.properties @@ -34,4 +34,4 @@ dhfs.objects.move-processor.threads=2 quarkus.log.category."com.usatiuk.dhfs".min-level=TRACE quarkus.log.category."com.usatiuk.dhfs".level=TRACE quarkus.http.insecure-requests=enabled -quarkus.http.ssl.client-auth=request +quarkus.http.ssl.client-auth=required