This is an automated email from the ASF dual-hosted git repository.
apkhmv pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/ignite-3.git
The following commit(s) were added to refs/heads/main by this push:
new 8b28bd1dcf IGNITE-19021 Support the directory deployment (#1931)
8b28bd1dcf is described below
commit 8b28bd1dcfe85e80abcd57ee6fb834594f366cde
Author: Vadim Pakhnushev <[email protected]>
AuthorDate: Wed Apr 26 10:49:47 2023 +0300
IGNITE-19021 Support the directory deployment (#1931)
---
.../cli/commands/unit/ItDeploymentUnitTest.java | 37 ++++--
.../internal/rest/ItGeneratedRestClientTest.java | 22 ++--
.../internal/cli/call/unit/DeployUnitCall.java | 68 ++++++-----
.../internal/cli/call/unit/DeployUnitClient.java | 100 ++++++++++++++++
.../cli/commands/unit/UnitDeployCommand.java | 2 +-
.../deployunit/DeployMessagingService.java | 8 +-
.../internal/deployunit/DeploymentManagerImpl.java | 45 +++++---
.../ignite/internal/deployunit/DeploymentUnit.java | 15 +--
.../internal/deployunit/FileDeployerService.java | 25 ++--
.../ignite/internal/deployunit/UnitMeta.java | 20 ++--
.../deployunit/key/UnitMetaSerializer.java | 64 +++++++----
.../deployunit/message/DeployUnitRequest.java | 15 +--
.../ignite/deployment/UnitMetaSerializerTest.java | 71 +++++-------
modules/rest-api/openapi/openapi.yaml | 6 +-
.../rest/api/deployment/DeploymentCodeApi.java | 4 +-
.../deployment/CompletedFileUploadSubscriber.java | 81 +++++++++++++
.../deployment/DeploymentManagementController.java | 40 ++-----
.../internal/deployment/ItDeploymentUnitTest.java | 128 ++++++++++++++-------
18 files changed, 504 insertions(+), 247 deletions(-)
diff --git
a/modules/cli/src/integrationTest/java/org/apache/ignite/internal/cli/commands/unit/ItDeploymentUnitTest.java
b/modules/cli/src/integrationTest/java/org/apache/ignite/internal/cli/commands/unit/ItDeploymentUnitTest.java
index 46549068fa..70868903de 100644
---
a/modules/cli/src/integrationTest/java/org/apache/ignite/internal/cli/commands/unit/ItDeploymentUnitTest.java
+++
b/modules/cli/src/integrationTest/java/org/apache/ignite/internal/cli/commands/unit/ItDeploymentUnitTest.java
@@ -21,8 +21,9 @@ import static org.junit.jupiter.api.Assertions.assertAll;
import java.io.IOException;
import java.nio.file.Files;
+import java.nio.file.Path;
import
org.apache.ignite.internal.cli.commands.CliCommandTestInitializedIntegrationBase;
-import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Disabled;
import org.junit.jupiter.api.DisplayName;
import org.junit.jupiter.api.Test;
@@ -31,18 +32,22 @@ import org.junit.jupiter.api.Test;
@Disabled("https://issues.apache.org/jira/browse/IGNITE-19139";)
public class ItDeploymentUnitTest extends
CliCommandTestInitializedIntegrationBase {
- private String path;
+ private String testFile;
- @BeforeEach
- void setUp() throws IOException {
- path = Files.createTempFile(WORK_DIR, "test",
"txt").toAbsolutePath().toString();
+ private Path testDirectory;
+
+ @BeforeAll
+ void beforeAll() throws IOException {
+ testDirectory = Files.createDirectory(WORK_DIR.resolve("test"));
+ testFile =
Files.createFile(testDirectory.resolve("test.txt")).toString();
+ Files.createFile(testDirectory.resolve("test2.txt"));
}
@Test
@DisplayName("Should deploy a unit with version")
void deploy() {
// When deploy with version
- execute("unit", "deploy", "test.unit.id.1", "--version", "1.0.0",
"--path", path);
+ execute("unit", "deploy", "test.unit.id.1", "--version", "1.0.0",
"--path", testFile);
// Then
assertAll(
@@ -56,7 +61,7 @@ public class ItDeploymentUnitTest extends
CliCommandTestInitializedIntegrationBa
@DisplayName("Should deploy a unit without version")
void deployWithoutVersion() {
// When deploy without version
- execute("unit", "deploy", "test.unit.id.2", "--path", path);
+ execute("unit", "deploy", "test.unit.id.2", "--path", testFile);
// Then
assertAll(
@@ -70,7 +75,7 @@ public class ItDeploymentUnitTest extends
CliCommandTestInitializedIntegrationBa
@DisplayName("Should undeploy a unit with version")
void undeploy() {
// When deploy
- execute("unit", "deploy", "test.unit.id.3", "--version", "1.0.0",
"--path", path);
+ execute("unit", "deploy", "test.unit.id.3", "--version", "1.0.0",
"--path", testFile);
// And undeploy
execute("unit", "undeploy", "test.unit.id.3", "--version", "1.0.0");
@@ -100,7 +105,7 @@ public class ItDeploymentUnitTest extends
CliCommandTestInitializedIntegrationBa
@DisplayName("Should display correct status after deploy")
void deployAndStatusCheck() {
// When undeploy non-existing unit
- execute("unit", "deploy", "test.unit.id.5", "--version", "1.0.0",
"--path", path);
+ execute("unit", "deploy", "test.unit.id.5", "--version", "1.0.0",
"--path", testFile);
// Then
assertAll(
@@ -118,4 +123,18 @@ public class ItDeploymentUnitTest extends
CliCommandTestInitializedIntegrationBa
() -> assertOutputContains("DEPLOYED")
);
}
+
+ @Test
+ @DisplayName("Should deploy a unit from directory")
+ void deployDirectory() {
+ // When deploy with version
+ execute("unit", "deploy", "test.unit.id.5", "--path",
testDirectory.toString());
+
+ // Then
+ assertAll(
+ this::assertExitCodeIsZero,
+ this::assertErrOutputIsEmpty,
+ () -> assertOutputContains("Done")
+ );
+ }
}
diff --git
a/modules/cli/src/integrationTest/java/org/apache/ignite/internal/rest/ItGeneratedRestClientTest.java
b/modules/cli/src/integrationTest/java/org/apache/ignite/internal/rest/ItGeneratedRestClientTest.java
index ec9a4bb4a9..f641c0af03 100644
---
a/modules/cli/src/integrationTest/java/org/apache/ignite/internal/rest/ItGeneratedRestClientTest.java
+++
b/modules/cli/src/integrationTest/java/org/apache/ignite/internal/rest/ItGeneratedRestClientTest.java
@@ -51,6 +51,7 @@ import java.util.stream.IntStream;
import org.apache.ignite.Ignite;
import org.apache.ignite.IgnitionManager;
import org.apache.ignite.InitParameters;
+import org.apache.ignite.internal.cli.call.unit.DeployUnitClient;
import org.apache.ignite.internal.cli.core.rest.ApiClientFactory;
import org.apache.ignite.internal.testframework.TestIgnitionManager;
import org.apache.ignite.internal.testframework.WorkDirectory;
@@ -100,6 +101,8 @@ public class ItGeneratedRestClientTest {
private final List<Ignite> clusterNodes = new ArrayList<>();
+ private ApiClient apiClient;
+
private ClusterConfigurationApi clusterConfigurationApi;
private NodeConfigurationApi nodeConfigurationApi;
@@ -157,15 +160,15 @@ public class ItGeneratedRestClientTest {
firstNodeName = clusterNodes.get(0).name();
- ApiClient client = clientFactory.getClient("http://localhost:"; +
BASE_REST_PORT);
+ apiClient = clientFactory.getClient("http://localhost:"; +
BASE_REST_PORT);
- clusterConfigurationApi = new ClusterConfigurationApi(client);
- nodeConfigurationApi = new NodeConfigurationApi(client);
- clusterManagementApi = new ClusterManagementApi(client);
- nodeManagementApi = new NodeManagementApi(client);
- topologyApi = new TopologyApi(client);
- nodeMetricApi = new NodeMetricApi(client);
- deploymentApi = new DeploymentApi(client);
+ clusterConfigurationApi = new ClusterConfigurationApi(apiClient);
+ nodeConfigurationApi = new NodeConfigurationApi(apiClient);
+ clusterManagementApi = new ClusterManagementApi(apiClient);
+ nodeManagementApi = new NodeManagementApi(apiClient);
+ topologyApi = new TopologyApi(apiClient);
+ nodeMetricApi = new NodeMetricApi(apiClient);
+ deploymentApi = new DeploymentApi(apiClient);
objectMapper = new ObjectMapper();
}
@@ -387,7 +390,8 @@ public class ItGeneratedRestClientTest {
void deployUndeployUnitSync() throws ApiException {
assertThat(deploymentApi.units(), empty());
- deploymentApi.deployUnit("test.unit.id", emptyFile(), "1.0.0");
+ // TODO https://issues.apache.org/jira/browse/IGNITE-19295
+ new DeployUnitClient(apiClient).deployUnit("test.unit.id",
List.of(emptyFile()), "1.0.0");
List<UnitStatus> units = deploymentApi.units();
assertThat(units, hasSize(1));
assertThat(units.get(0).getId(), equalTo("test.unit.id"));
diff --git
a/modules/cli/src/main/java/org/apache/ignite/internal/cli/call/unit/DeployUnitCall.java
b/modules/cli/src/main/java/org/apache/ignite/internal/cli/call/unit/DeployUnitCall.java
index 8d8a94547a..41d38ca5e8 100644
---
a/modules/cli/src/main/java/org/apache/ignite/internal/cli/call/unit/DeployUnitCall.java
+++
b/modules/cli/src/main/java/org/apache/ignite/internal/cli/call/unit/DeployUnitCall.java
@@ -21,7 +21,13 @@ import static
java.util.concurrent.CompletableFuture.completedFuture;
import java.io.File;
import java.io.FileNotFoundException;
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.List;
import java.util.concurrent.CompletableFuture;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
import okhttp3.Call;
import org.apache.ignite.internal.cli.core.call.AsyncCall;
import org.apache.ignite.internal.cli.core.call.CallOutput;
@@ -33,7 +39,7 @@ import
org.apache.ignite.internal.cli.core.repl.registry.UnitsRegistry;
import org.apache.ignite.internal.cli.core.rest.ApiClientFactory;
import org.apache.ignite.internal.cli.core.style.component.MessageUiComponent;
import org.apache.ignite.internal.cli.core.style.element.UiElements;
-import org.apache.ignite.rest.client.api.DeploymentApi;
+import org.apache.ignite.rest.client.invoker.ApiClient;
import org.apache.ignite.rest.client.invoker.ApiException;
/** Call to deploy a unit. */
@@ -53,36 +59,42 @@ public class DeployUnitCall implements
AsyncCall<DeployUnitCallInput, String> {
@Override
public CompletableFuture<CallOutput<String>> execute(DeployUnitCallInput
input) {
- try {
- DeploymentApi api = new
DeploymentApi(clientFactory.getClient(input.clusterUrl()));
+ ApiClient apiClient = clientFactory.getClient(input.clusterUrl());
+ DeployUnitClient api = new DeployUnitClient(apiClient);
- File file = input.path().toFile();
- if (!file.exists()) {
- return completedFuture(DefaultCallOutput.failure(new
FileNotFoundException(input.path().toString())));
- }
+ Path path = input.path();
+ if (Files.notExists(path)) {
+ return completedFuture(DefaultCallOutput.failure(new
FileNotFoundException(path.toString())));
+ }
+ List<File> files;
+ try (Stream<Path> stream = Files.walk(path, 1)) {
+ files = stream
+ .filter(Files::isRegularFile)
+ .map(Path::toFile)
+ .collect(Collectors.toList());
+ } catch (IOException e) {
+ return completedFuture(DefaultCallOutput.failure(e));
+ }
- TrackingCallback<Boolean> callback = new
TrackingCallback<>(tracker);
- String ver = input.version() == null ? "" : input.version();
- Call call = api.deployUnitAsync(input.id(), file, ver, callback);
+ TrackingCallback<Boolean> callback = new TrackingCallback<>(tracker);
+ String ver = input.version() == null ? "" : input.version();
+ Call call = api.deployUnitAsync(input.id(), files, ver, callback);
- return CompletableFuture.supplyAsync(() -> {
- try {
- callback.awaitDone();
- } catch (InterruptedException e) {
- return DefaultCallOutput.failure(e);
- }
- if (call.isCanceled()) {
- return DefaultCallOutput.failure(new
RuntimeException("Unit deployment process was canceled"));
- } else if (callback.exception() != null) {
- return handleException(callback.exception(), input);
- } else {
- unitsRegistry.refresh();
- return
DefaultCallOutput.success(MessageUiComponent.from(UiElements.done()).render());
- }
- });
- } catch (ApiException e) {
- return completedFuture(DefaultCallOutput.failure(new
IgniteCliApiException(e, input.clusterUrl())));
- }
+ return CompletableFuture.supplyAsync(() -> {
+ try {
+ callback.awaitDone();
+ } catch (InterruptedException e) {
+ return DefaultCallOutput.failure(e);
+ }
+ if (call.isCanceled()) {
+ return DefaultCallOutput.failure(new RuntimeException("Unit
deployment process was canceled"));
+ } else if (callback.exception() != null) {
+ return handleException(callback.exception(), input);
+ } else {
+ unitsRegistry.refresh();
+ return
DefaultCallOutput.success(MessageUiComponent.from(UiElements.done()).render());
+ }
+ });
}
private static CallOutput<String> handleException(Exception exception,
DeployUnitCallInput input) {
diff --git
a/modules/cli/src/main/java/org/apache/ignite/internal/cli/call/unit/DeployUnitClient.java
b/modules/cli/src/main/java/org/apache/ignite/internal/cli/call/unit/DeployUnitClient.java
new file mode 100644
index 0000000000..ecf2886e13
--- /dev/null
+++
b/modules/cli/src/main/java/org/apache/ignite/internal/cli/call/unit/DeployUnitClient.java
@@ -0,0 +1,100 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.cli.call.unit;
+
+import java.io.File;
+import java.util.List;
+import okhttp3.Call;
+import okhttp3.MediaType;
+import okhttp3.MultipartBody;
+import okhttp3.Request;
+import okhttp3.RequestBody;
+import org.apache.ignite.rest.client.invoker.ApiCallback;
+import org.apache.ignite.rest.client.invoker.ApiClient;
+import org.apache.ignite.rest.client.invoker.ApiException;
+import org.apache.ignite.rest.client.invoker.ApiResponse;
+import org.apache.ignite.rest.client.invoker.ProgressRequestBody;
+
+/**
+ * Temporary class for calling REST with list of files until underlying issue
in the openapi-generator is fixed.
+ * TODO https://issues.apache.org/jira/browse/IGNITE-19295
+ */
+public class DeployUnitClient {
+ private final ApiClient apiClient;
+
+ public DeployUnitClient(ApiClient apiClient) {
+ this.apiClient = apiClient;
+ }
+
+ /**
+ * Deploy unit.
+ *
+ * @param unitId The ID of the deployment unit.
+ * @param unitContent The code to deploy.
+ * @param unitVersion The version of the deployment unit.
+ * @return {@code true} if the call succeeded.
+ * @throws ApiException if fail to call.
+ */
+ public Boolean deployUnit(String unitId, List<File> unitContent, String
unitVersion) throws ApiException {
+ Call call = deployUnitCall(unitId, unitContent, unitVersion, null);
+ ApiResponse<Boolean> response = apiClient.execute(call, Boolean.class);
+ return response.getData();
+ }
+
+ /**
+ * Deploy unit asynchronously.
+ *
+ * @param unitId The ID of the deployment unit.
+ * @param unitContent The code to deploy.
+ * @param unitVersion The version of the deployment unit.
+ * @return Request call.
+ */
+ public Call deployUnitAsync(String unitId, List<File> unitContent, String
unitVersion, ApiCallback<Boolean> callback) {
+ Call call = deployUnitCall(unitId, unitContent, unitVersion, callback);
+ apiClient.executeAsync(call, Boolean.class, callback);
+ return call;
+ }
+
+ private Call deployUnitCall(String unitId, List<File> unitContent, String
unitVersion, ApiCallback<Boolean> callback) {
+ String url = apiClient.getBasePath() +
"/management/v1/deployment/units";
+
+ MultipartBody.Builder mpBuilder = new
MultipartBody.Builder().setType(MultipartBody.FORM);
+ mpBuilder.addFormDataPart("unitId", unitId);
+ mpBuilder.addFormDataPart("unitVersion", unitVersion);
+ for (File file : unitContent) {
+ RequestBody requestBody = RequestBody.create(file,
MediaType.parse("application/octet-stream"));
+ mpBuilder.addFormDataPart("unitContent", file.getName(),
requestBody);
+ }
+ MultipartBody body = mpBuilder.build();
+
+ Request.Builder reqBuilder = new Request.Builder()
+ .url(url)
+ .header("Accept", "application/json")
+ .header("Content-Type", "multipart/form-data");
+
+ if (callback != null) {
+ ProgressRequestBody progressRequestBody = new
ProgressRequestBody(body, callback);
+ reqBuilder.tag(callback)
+ .post(progressRequestBody);
+ } else {
+ reqBuilder.post(body);
+ }
+
+ return apiClient.getHttpClient().newCall(reqBuilder.build());
+ }
+}
diff --git
a/modules/cli/src/main/java/org/apache/ignite/internal/cli/commands/unit/UnitDeployCommand.java
b/modules/cli/src/main/java/org/apache/ignite/internal/cli/commands/unit/UnitDeployCommand.java
index 2ebf35d163..282be0873a 100644
---
a/modules/cli/src/main/java/org/apache/ignite/internal/cli/commands/unit/UnitDeployCommand.java
+++
b/modules/cli/src/main/java/org/apache/ignite/internal/cli/commands/unit/UnitDeployCommand.java
@@ -42,7 +42,7 @@ import picocli.CommandLine.ParameterException;
import picocli.CommandLine.Parameters;
/** Command to deploy a unit. */
-@Command(name = "deploy", description = "Deploys a unit")
+@Command(name = "deploy", description = "Deploys a unit from file or a
directory (non-recursively)")
public class UnitDeployCommand extends BaseCommand implements
Callable<Integer> {
@Mixin
diff --git
a/modules/code-deployment/src/main/java/org/apache/ignite/internal/deployunit/DeployMessagingService.java
b/modules/code-deployment/src/main/java/org/apache/ignite/internal/deployunit/DeployMessagingService.java
index 7461d2772c..0f173c4377 100644
---
a/modules/code-deployment/src/main/java/org/apache/ignite/internal/deployunit/DeployMessagingService.java
+++
b/modules/code-deployment/src/main/java/org/apache/ignite/internal/deployunit/DeployMessagingService.java
@@ -112,13 +112,11 @@ public class DeployMessagingService {
*
* @param id Deployment unit identifier.
* @param version Deployment unit version.
- * @param unitName Deployment unit file name.
- * @param unitContent Deployment unit file content.
+ * @param unitContent Deployment unit file names and content.
* @return Future with deployment result.
*/
- public CompletableFuture<List<String>> startDeployAsyncToCmg(String id,
Version version, String unitName, byte[] unitContent) {
+ public CompletableFuture<List<String>> startDeployAsyncToCmg(String id,
Version version, Map<String, byte[]> unitContent) {
DeployUnitRequest request = DeployUnitRequestImpl.builder()
- .unitName(unitName)
.id(id)
.version(version.render())
.unitContent(unitContent)
@@ -214,7 +212,7 @@ public class DeployMessagingService {
String id = executeRequest.id();
String version = executeRequest.version();
tracker.track(id, Version.parseVersion(version),
- deployerService.deploy(id, version, executeRequest.unitName(),
executeRequest.unitContent())
+ deployerService.deploy(id, version,
executeRequest.unitContent())
.thenCompose(success ->
clusterService.messagingService().respond(
senderConsistentId,
DEPLOYMENT_CHANNEL,
diff --git
a/modules/code-deployment/src/main/java/org/apache/ignite/internal/deployunit/DeploymentManagerImpl.java
b/modules/code-deployment/src/main/java/org/apache/ignite/internal/deployunit/DeploymentManagerImpl.java
index c1cd919bbe..91c4289dc4 100644
---
a/modules/code-deployment/src/main/java/org/apache/ignite/internal/deployunit/DeploymentManagerImpl.java
+++
b/modules/code-deployment/src/main/java/org/apache/ignite/internal/deployunit/DeploymentManagerImpl.java
@@ -18,17 +18,23 @@
package org.apache.ignite.internal.deployunit;
import static java.util.concurrent.CompletableFuture.allOf;
+import static java.util.concurrent.CompletableFuture.completedFuture;
+import static java.util.concurrent.CompletableFuture.failedFuture;
import static
org.apache.ignite.internal.rest.api.deployment.DeploymentStatus.DEPLOYED;
import static
org.apache.ignite.internal.rest.api.deployment.DeploymentStatus.OBSOLETE;
import static
org.apache.ignite.internal.rest.api.deployment.DeploymentStatus.REMOVING;
import static
org.apache.ignite.internal.rest.api.deployment.DeploymentStatus.UPLOADING;
import java.io.IOException;
+import java.io.InputStream;
import java.nio.file.Path;
import java.util.Collections;
import java.util.List;
+import java.util.Map;
+import java.util.Map.Entry;
import java.util.Objects;
import java.util.concurrent.CompletableFuture;
+import java.util.stream.Collectors;
import
org.apache.ignite.internal.cluster.management.ClusterManagementGroupManager;
import
org.apache.ignite.internal.deployunit.configuration.DeploymentConfiguration;
import
org.apache.ignite.internal.deployunit.exception.DeploymentUnitAlreadyExistsException;
@@ -122,31 +128,31 @@ public class DeploymentManagerImpl implements
IgniteDeployment {
Objects.requireNonNull(version);
Objects.requireNonNull(deploymentUnit);
- UnitMeta meta = new UnitMeta(id, version, deploymentUnit.name(),
UPLOADING, Collections.emptyList());
-
- byte[] unitContent;
- try {
- unitContent = deploymentUnit.content().readAllBytes();
- } catch (IOException e) {
- LOG.error("Error to read deployment unit content", e);
- return CompletableFuture.failedFuture(new
DeploymentUnitReadException(e));
- }
+ List<String> fileNames =
List.copyOf(deploymentUnit.content().keySet());
+ UnitMeta meta = new UnitMeta(id, version, fileNames, UPLOADING,
Collections.emptyList());
return metastore.putIfNotExist(id, version, meta)
.thenCompose(success -> {
if (success) {
- return tracker.track(id, version, deployer.deploy(id,
version.render(), deploymentUnit.name(), unitContent)
+ Map<String, byte[]> unitContent;
+ try {
+ unitContent =
deploymentUnit.content().entrySet().stream()
+ .collect(Collectors.toMap(Entry::getKey,
entry -> readContent(entry.getValue())));
+ } catch (DeploymentUnitReadException e) {
+ return failedFuture(e);
+ }
+ return tracker.track(id, version, deployer.deploy(id,
version.render(), unitContent)
.thenCompose(deployed -> {
if (deployed) {
return metastore.updateMeta(id,
version,
unitMeta -> unitMeta
.addConsistentId(clusterService.topologyService().localMember().name()));
}
- return
CompletableFuture.completedFuture(false);
+ return completedFuture(false);
})
.thenApply(completed -> {
if (completed) {
- messaging.startDeployAsyncToCmg(id,
version, deploymentUnit.name(), unitContent)
+ messaging.startDeployAsyncToCmg(id,
version, unitContent)
.thenAccept(ids ->
metastore.updateMeta(id, version, unitMeta -> {
for (String consistentId :
ids) {
unitMeta.addConsistentId(consistentId);
@@ -163,7 +169,7 @@ public class DeploymentManagerImpl implements
IgniteDeployment {
}
LOG.warn("Failed to deploy meta of unit " + id + ":" +
version + " to metastore. "
+ "Already exists.");
- return CompletableFuture.failedFuture(
+ return failedFuture(
new DeploymentUnitAlreadyExistsException(id,
"Unit " + id + ":" + version + "
already exists"));
}
@@ -183,13 +189,13 @@ public class DeploymentManagerImpl implements
IgniteDeployment {
//TODO: Check unit usages here. If unit used in
compute task we cannot just remove it.
return metastore.updateMeta(id, version, true, meta ->
meta.updateStatus(REMOVING));
}
- return CompletableFuture.completedFuture(false);
+ return completedFuture(false);
})
.thenCompose(success -> {
if (success) {
return cmgManager.logicalTopology();
}
- return CompletableFuture.failedFuture(new
DeploymentUnitNotFoundException(
+ return failedFuture(new DeploymentUnitNotFoundException(
"Unit " + id + " with version " + version + "
doesn't exist"));
}).thenCompose(logicalTopologySnapshot -> allOf(
logicalTopologySnapshot.nodes().stream()
@@ -262,4 +268,13 @@ public class DeploymentManagerImpl implements
IgniteDeployment {
throw new IllegalArgumentException("Id is blank");
}
}
+
+ private static byte[] readContent(InputStream inputStream) {
+ try (inputStream) {
+ return inputStream.readAllBytes();
+ } catch (IOException e) {
+ LOG.error("Error reading deployment unit content", e);
+ throw new DeploymentUnitReadException(e);
+ }
+ }
}
diff --git
a/modules/code-deployment/src/main/java/org/apache/ignite/internal/deployunit/DeploymentUnit.java
b/modules/code-deployment/src/main/java/org/apache/ignite/internal/deployunit/DeploymentUnit.java
index 4e206eb0ae..3c4bddd4bf 100644
---
a/modules/code-deployment/src/main/java/org/apache/ignite/internal/deployunit/DeploymentUnit.java
+++
b/modules/code-deployment/src/main/java/org/apache/ignite/internal/deployunit/DeploymentUnit.java
@@ -18,22 +18,17 @@
package org.apache.ignite.internal.deployunit;
import java.io.InputStream;
+import java.util.Map;
/**
* Deployment unit interface.
*/
+@FunctionalInterface
public interface DeploymentUnit {
/**
- * Unit name.
+ * Deployment unit content - a map from file name to input stream.
*
- * @return Name of deployment unit.
+ * @return Deployment unit content.
*/
- String name();
-
- /**
- * Input stream with deployment unit content.
- *
- * @return input stream with deployment unit content.
- */
- InputStream content();
+ Map<String, InputStream> content();
}
diff --git
a/modules/code-deployment/src/main/java/org/apache/ignite/internal/deployunit/FileDeployerService.java
b/modules/code-deployment/src/main/java/org/apache/ignite/internal/deployunit/FileDeployerService.java
index fe4c7dfab3..be8adb7743 100644
---
a/modules/code-deployment/src/main/java/org/apache/ignite/internal/deployunit/FileDeployerService.java
+++
b/modules/code-deployment/src/main/java/org/apache/ignite/internal/deployunit/FileDeployerService.java
@@ -26,6 +26,8 @@ import static
java.nio.file.StandardOpenOption.TRUNCATE_EXISTING;
import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Path;
+import java.util.Map;
+import java.util.Map.Entry;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
@@ -62,24 +64,25 @@ public class FileDeployerService {
*
* @param id Deploy unit identifier.
* @param version Deploy unit version.
- * @param unitName Deploy unit file name.
- * @param unitContent Deploy unit content.
+ * @param unitFileContent Map of deploy unit file names to file content.
* @return Future with deploy result.
*/
- public CompletableFuture<Boolean> deploy(String id, String version, String
unitName, byte[] unitContent) {
+ public CompletableFuture<Boolean> deploy(String id, String version,
Map<String, byte[]> unitFileContent) {
return CompletableFuture.supplyAsync(() -> {
try {
- Path unitPath = unitsFolder
+ Path unitFolder = unitsFolder
.resolve(id)
- .resolve(version)
- .resolve(unitName);
-
- Path unitPathTmp =
unitPath.resolveSibling(unitPath.getFileName() + TMP_SUFFIX);
+ .resolve(version);
- Files.createDirectories(unitPathTmp.getParent());
+ Files.createDirectories(unitFolder);
- Files.write(unitPathTmp, unitContent, CREATE, SYNC,
TRUNCATE_EXISTING);
- Files.move(unitPathTmp, unitPath, ATOMIC_MOVE,
REPLACE_EXISTING);
+ for (Entry<String, byte[]> entry : unitFileContent.entrySet())
{
+ String fileName = entry.getKey();
+ Path unitPath = unitFolder.resolve(fileName);
+ Path unitPathTmp = unitFolder.resolve(fileName +
TMP_SUFFIX);
+ Files.write(unitPathTmp, entry.getValue(), CREATE, SYNC,
TRUNCATE_EXISTING);
+ Files.move(unitPathTmp, unitPath, ATOMIC_MOVE,
REPLACE_EXISTING);
+ }
return true;
} catch (IOException e) {
LOG.error("Failed to deploy unit " + id + ":" + version, e);
diff --git
a/modules/code-deployment/src/main/java/org/apache/ignite/internal/deployunit/UnitMeta.java
b/modules/code-deployment/src/main/java/org/apache/ignite/internal/deployunit/UnitMeta.java
index c34ec09928..391deb36b8 100644
---
a/modules/code-deployment/src/main/java/org/apache/ignite/internal/deployunit/UnitMeta.java
+++
b/modules/code-deployment/src/main/java/org/apache/ignite/internal/deployunit/UnitMeta.java
@@ -37,9 +37,9 @@ public class UnitMeta {
private final Version version;
/**
- * Unit name.
+ * Unit file names.
*/
- private final String name;
+ private final List<String> fileNames;
/**
* Deployment status.
@@ -56,13 +56,13 @@ public class UnitMeta {
*
* @param id Unit identifier.
* @param version Unit version.
- * @param name Unit name.
+ * @param fileNames Unit file names.
* @param consistentIdLocation Consistent ids of nodes where unit deployed.
*/
- public UnitMeta(String id, Version version, String name, DeploymentStatus
status, List<String> consistentIdLocation) {
+ public UnitMeta(String id, Version version, List<String> fileNames,
DeploymentStatus status, List<String> consistentIdLocation) {
this.id = id;
this.version = version;
- this.name = name;
+ this.fileNames = fileNames;
this.status = status;
this.consistentIdLocation.addAll(consistentIdLocation);
}
@@ -90,8 +90,8 @@ public class UnitMeta {
*
* @return name of deployment unit.
*/
- public String name() {
- return name;
+ public List<String> fileNames() {
+ return fileNames;
}
/**
@@ -137,7 +137,7 @@ public class UnitMeta {
if (version != null ? !version.equals(meta.version) : meta.version !=
null) {
return false;
}
- if (name != null ? !name.equals(meta.name) : meta.name != null) {
+ if (fileNames != null ? !fileNames.equals(meta.fileNames) :
meta.fileNames != null) {
return false;
}
if (status != meta.status) {
@@ -150,7 +150,7 @@ public class UnitMeta {
public int hashCode() {
int result = id != null ? id.hashCode() : 0;
result = 31 * result + (version != null ? version.hashCode() : 0);
- result = 31 * result + (name != null ? name.hashCode() : 0);
+ result = 31 * result + (fileNames != null ? fileNames.hashCode() : 0);
result = 31 * result + (status != null ? status.hashCode() : 0);
result = 31 * result + consistentIdLocation.hashCode();
return result;
@@ -161,7 +161,7 @@ public class UnitMeta {
return "UnitMeta{"
+ "id='" + id + '\''
+ ", version=" + version
- + ", name='" + name + '\''
+ + ", fileNames=" + String.join(", ", fileNames)
+ ", status=" + status
+ ", consistentIdLocation=" + String.join(", ",
consistentIdLocation)
+ '}';
diff --git
a/modules/code-deployment/src/main/java/org/apache/ignite/internal/deployunit/key/UnitMetaSerializer.java
b/modules/code-deployment/src/main/java/org/apache/ignite/internal/deployunit/key/UnitMetaSerializer.java
index e1d927e960..669b061073 100644
---
a/modules/code-deployment/src/main/java/org/apache/ignite/internal/deployunit/key/UnitMetaSerializer.java
+++
b/modules/code-deployment/src/main/java/org/apache/ignite/internal/deployunit/key/UnitMetaSerializer.java
@@ -19,11 +19,11 @@ package org.apache.ignite.internal.deployunit.key;
import static java.nio.charset.StandardCharsets.UTF_8;
-import java.util.ArrayList;
+import java.util.Arrays;
import java.util.Base64;
-import java.util.Base64.Decoder;
-import java.util.Base64.Encoder;
+import java.util.Collections;
import java.util.List;
+import java.util.stream.Collectors;
import org.apache.ignite.internal.deployunit.UnitMeta;
import org.apache.ignite.internal.deployunit.version.Version;
import org.apache.ignite.internal.rest.api.deployment.DeploymentStatus;
@@ -34,6 +34,8 @@ import
org.apache.ignite.internal.rest.api.deployment.DeploymentStatus;
public final class UnitMetaSerializer {
private static final String SEPARATOR = ";";
+ private static final String LIST_SEPARATOR = ":";
+
/**
* Constructor.
*/
@@ -52,21 +54,13 @@ public final class UnitMetaSerializer {
appendWithEncoding(sb, meta.id());
appendWithEncoding(sb, meta.version().render());
- appendWithEncoding(sb, meta.name());
+ appendWithEncoding(sb, meta.fileNames());
appendWithEncoding(sb, meta.status().name());
-
- for (String id : meta.consistentIdLocation()) {
- appendWithEncoding(sb, id);
- }
+ appendWithEncoding(sb, meta.consistentIdLocation());
return sb.toString().getBytes(UTF_8);
}
- private static void appendWithEncoding(StringBuilder sb, String content) {
- Encoder encoder = Base64.getEncoder();
- sb.append(new String(encoder.encode(content.getBytes(UTF_8)),
UTF_8)).append(SEPARATOR);
- }
-
/**
* Deserialize byte array to unit meta.
*
@@ -75,21 +69,45 @@ public final class UnitMetaSerializer {
*/
public static UnitMeta deserialize(byte[] bytes) {
String s = new String(bytes, UTF_8);
- String[] split = s.split(SEPARATOR);
+ String[] split = s.split(SEPARATOR, -1);
+
+ String id = decode(split[0]);
+ String version = decode(split[1]);
+ List<String> fileNames = deserializeList(split[2]);
- Decoder decoder = Base64.getDecoder();
+ DeploymentStatus status = DeploymentStatus.valueOf(decode(split[3]));
- String id = new String(decoder.decode(split[0]), UTF_8);
- String version = new String(decoder.decode(split[1]), UTF_8);
- String unitName = new String(decoder.decode(split[2]), UTF_8);
+ List<String> ids = deserializeList(split[4]);
- DeploymentStatus status = DeploymentStatus.valueOf(new
String(decoder.decode(split[3]), UTF_8));
+ return new UnitMeta(id, Version.parseVersion(version), fileNames,
status, ids);
+ }
- List<String> ids = new ArrayList<>();
- for (int i = 4; i < split.length; i++) {
- ids.add(new String(decoder.decode(split[i]), UTF_8));
+ private static void appendWithEncoding(StringBuilder sb, String content) {
+ sb.append(encode(content)).append(SEPARATOR);
+ }
+
+ private static void appendWithEncoding(StringBuilder sb, List<String>
content) {
+ String list = content.stream()
+ .map(UnitMetaSerializer::encode)
+ .collect(Collectors.joining(LIST_SEPARATOR));
+ sb.append(list).append(SEPARATOR);
+ }
+
+ private static List<String> deserializeList(String data) {
+ if (data.isEmpty()) {
+ return Collections.emptyList();
}
- return new UnitMeta(id, Version.parseVersion(version), unitName,
status, ids);
+ return Arrays.stream(data.split(LIST_SEPARATOR))
+ .map(UnitMetaSerializer::decode)
+ .collect(Collectors.toList());
+ }
+
+ private static String encode(String s) {
+ return new String(Base64.getEncoder().encode(s.getBytes(UTF_8)),
UTF_8);
+ }
+
+ private static String decode(String s) {
+ return new String(Base64.getDecoder().decode(s), UTF_8);
}
}
diff --git
a/modules/code-deployment/src/main/java/org/apache/ignite/internal/deployunit/message/DeployUnitRequest.java
b/modules/code-deployment/src/main/java/org/apache/ignite/internal/deployunit/message/DeployUnitRequest.java
index ddf16bf6d7..7e88719300 100644
---
a/modules/code-deployment/src/main/java/org/apache/ignite/internal/deployunit/message/DeployUnitRequest.java
+++
b/modules/code-deployment/src/main/java/org/apache/ignite/internal/deployunit/message/DeployUnitRequest.java
@@ -17,6 +17,7 @@
package org.apache.ignite.internal.deployunit.message;
+import java.util.Map;
import org.apache.ignite.network.NetworkMessage;
import org.apache.ignite.network.annotations.Transferable;
@@ -40,18 +41,10 @@ public interface DeployUnitRequest extends NetworkMessage {
String version();
/**
- * Returns name of deployment unit.
+ * Returns map from file names of deployment unit to their content.
*
- * @return name of deployment unit.
+ * @return map from file names of deployment unit to their content.
*/
- String unitName();
-
- /**
- * Returns content of deployment unit.
- *
- * @return content of deployment unit.
- */
-
- byte[] unitContent();
+ Map<String, byte[]> unitContent();
}
diff --git
a/modules/code-deployment/src/test/java/org/apache/ignite/deployment/UnitMetaSerializerTest.java
b/modules/code-deployment/src/test/java/org/apache/ignite/deployment/UnitMetaSerializerTest.java
index 866dc30cc6..8d522044cf 100644
---
a/modules/code-deployment/src/test/java/org/apache/ignite/deployment/UnitMetaSerializerTest.java
+++
b/modules/code-deployment/src/test/java/org/apache/ignite/deployment/UnitMetaSerializerTest.java
@@ -20,64 +20,51 @@ package org.apache.ignite.deployment;
import static
org.apache.ignite.internal.deployunit.key.UnitMetaSerializer.deserialize;
import static
org.apache.ignite.internal.deployunit.key.UnitMetaSerializer.serialize;
import static
org.apache.ignite.internal.rest.api.deployment.DeploymentStatus.UPLOADING;
+import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.Matchers.is;
+import static org.junit.jupiter.params.provider.Arguments.arguments;
import java.util.Arrays;
import java.util.Collections;
+import java.util.List;
import org.apache.ignite.internal.deployunit.UnitMeta;
import org.apache.ignite.internal.deployunit.key.UnitMetaSerializer;
import org.apache.ignite.internal.deployunit.version.Version;
-import org.hamcrest.MatcherAssert;
-import org.junit.jupiter.api.Test;
+import org.apache.ignite.internal.rest.api.deployment.DeploymentStatus;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
/**
* Test for {@link UnitMetaSerializer}.
*/
public class UnitMetaSerializerTest {
- @Test
- public void testSerializeDeserializeLatest() {
- UnitMeta meta = new UnitMeta("id", Version.LATEST, "unitName",
UPLOADING, Arrays.asList("id1", "id2"));
-
- byte[] serialize = serialize(meta);
-
- MatcherAssert.assertThat(deserialize(serialize), is(meta));
- }
-
- @Test
- public void testSerializeDeserializeUnit() {
- UnitMeta meta = new UnitMeta("id", Version.parseVersion("3.0.0"),
"unitName", UPLOADING, Arrays.asList("id1", "id2"));
-
- byte[] serialize = serialize(meta);
-
- MatcherAssert.assertThat(deserialize(serialize), is(meta));
- }
-
- @Test
- public void testSerializeDeserializeUnitIncompleteVersion() {
- UnitMeta meta = new UnitMeta("id", Version.parseVersion("3.0"),
"unitName", UPLOADING, Arrays.asList("id1", "id2"));
-
- byte[] serialize = serialize(meta);
-
- MatcherAssert.assertThat(deserialize(serialize), is(meta));
- }
-
- @Test
- public void testSerializeDeserializeUnitEmptyConsistentId() {
- UnitMeta meta = new UnitMeta("id", Version.parseVersion("3.0.0"),
"unitName", UPLOADING, Collections.emptyList());
-
- byte[] serialize = serialize(meta);
-
- UnitMeta deserialize = deserialize(serialize);
- MatcherAssert.assertThat(deserialize, is(meta));
+ private static List<Arguments> metaProvider() {
+ return List.of(
+ arguments("id", Version.LATEST, List.of("fileName"),
UPLOADING, Arrays.asList("id1", "id2")),
+ arguments("id", Version.LATEST, List.of("fileName1",
"fileName2"), UPLOADING, Arrays.asList("id1", "id2")),
+ arguments("id", Version.parseVersion("3.0.0"),
List.of("fileName"), UPLOADING, Arrays.asList("id1", "id2")),
+ arguments("id", Version.parseVersion("3.0"),
List.of("fileName"), UPLOADING, Arrays.asList("id1", "id2")),
+ arguments("id", Version.parseVersion("3.0.0"),
List.of("fileName"), UPLOADING, Collections.emptyList()),
+ arguments("id", Version.parseVersion("3.0.0"),
List.of("fileName1", "fileName2"), UPLOADING, Collections.emptyList()),
+ arguments("id;", Version.parseVersion("3.0.0"),
List.of("fileName;"), UPLOADING, Collections.emptyList()),
+ arguments("id;", Version.parseVersion("3.0.0"),
List.of("fileName1:;", "fileName2"), UPLOADING, Collections.emptyList())
+ );
}
- @Test
- public void testSerializeDeserializeWithSeparatorCharInIdName() {
- UnitMeta meta = new UnitMeta("id;", Version.parseVersion("3.0.0"),
"unitName;", UPLOADING, Collections.emptyList());
+ @ParameterizedTest
+ @MethodSource("metaProvider")
+ public void testSerializeDeserialize(
+ String id,
+ Version version,
+ List<String> fileNames,
+ DeploymentStatus status,
+ List<String> consistentIdLocation
+ ) {
+ UnitMeta meta = new UnitMeta(id, version, fileNames, status,
consistentIdLocation);
byte[] serialize = serialize(meta);
- UnitMeta deserialize = deserialize(serialize);
- MatcherAssert.assertThat(deserialize, is(meta));
+ assertThat(deserialize(serialize), is(meta));
}
}
diff --git a/modules/rest-api/openapi/openapi.yaml
b/modules/rest-api/openapi/openapi.yaml
index 86fbe21ebf..f7dd0156c9 100644
--- a/modules/rest-api/openapi/openapi.yaml
+++ b/modules/rest-api/openapi/openapi.yaml
@@ -365,9 +365,11 @@ paths:
type: string
description: The version of the deployment unit.
unitContent:
- type: string
+ type: array
description: The code to deploy.
- format: binary
+ items:
+ type: string
+ format: binary
required: true
responses:
"200":
diff --git
a/modules/rest-api/src/main/java/org/apache/ignite/internal/rest/api/deployment/DeploymentCodeApi.java
b/modules/rest-api/src/main/java/org/apache/ignite/internal/rest/api/deployment/DeploymentCodeApi.java
index 5abfafb69b..387d139da0 100644
---
a/modules/rest-api/src/main/java/org/apache/ignite/internal/rest/api/deployment/DeploymentCodeApi.java
+++
b/modules/rest-api/src/main/java/org/apache/ignite/internal/rest/api/deployment/DeploymentCodeApi.java
@@ -39,6 +39,7 @@ import io.swagger.v3.oas.annotations.tags.Tag;
import java.util.Collection;
import java.util.concurrent.CompletableFuture;
import org.apache.ignite.internal.rest.api.Problem;
+import org.reactivestreams.Publisher;
/**
* REST endpoint allows to deployment code service.
@@ -70,7 +71,8 @@ public interface DeploymentCodeApi {
description = "The version of the deployment unit.")
String unitVersion,
@Schema(name = "unitContent",
requiredMode = RequiredMode.REQUIRED,
- description = "The code to deploy.") CompletedFileUpload
unitContent);
+ description = "The code to deploy.")
Publisher<CompletedFileUpload> unitContent
+ );
/**
* Undeploy unit REST method.
diff --git
a/modules/rest/src/main/java/org/apache/ignite/internal/rest/deployment/CompletedFileUploadSubscriber.java
b/modules/rest/src/main/java/org/apache/ignite/internal/rest/deployment/CompletedFileUploadSubscriber.java
new file mode 100644
index 0000000000..4eb5fba651
--- /dev/null
+++
b/modules/rest/src/main/java/org/apache/ignite/internal/rest/deployment/CompletedFileUploadSubscriber.java
@@ -0,0 +1,81 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.ignite.internal.rest.deployment;
+
+import io.micronaut.http.multipart.CompletedFileUpload;
+import java.io.IOException;
+import java.io.InputStream;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import org.apache.ignite.internal.deployunit.DeploymentUnit;
+import org.reactivestreams.Subscriber;
+import org.reactivestreams.Subscription;
+
+/**
+ * Implementation of {@link Subscriber} based on {@link CompletedFileUpload}
which will collect uploaded files to the
+ * {@link DeploymentUnit}.
+ */
+class CompletedFileUploadSubscriber implements Subscriber<CompletedFileUpload>
{
+ private final CompletableFuture<DeploymentUnit> result;
+
+ private final Map<String, InputStream> content = new HashMap<>();
+
+ private IOException ex;
+
+ /**
+ * Constructor.
+ *
+ * @param result Result future.
+ */
+ CompletedFileUploadSubscriber(CompletableFuture<DeploymentUnit> result) {
+ this.result = result;
+ }
+
+ @Override
+ public void onSubscribe(Subscription subscription) {
+ subscription.request(Long.MAX_VALUE);
+ }
+
+ @Override
+ public void onNext(CompletedFileUpload item) {
+ try {
+ content.put(item.getFilename(), item.getInputStream());
+ } catch (IOException e) {
+ if (ex != null) {
+ ex.addSuppressed(e);
+ } else {
+ ex = e;
+ }
+ }
+ }
+
+ @Override
+ public void onError(Throwable throwable) {
+ result.completeExceptionally(throwable);
+ }
+
+ @Override
+ public void onComplete() {
+ if (ex != null) {
+ result.completeExceptionally(ex);
+ } else {
+ result.complete(() -> content);
+ }
+ }
+}
diff --git
a/modules/rest/src/main/java/org/apache/ignite/internal/rest/deployment/DeploymentManagementController.java
b/modules/rest/src/main/java/org/apache/ignite/internal/rest/deployment/DeploymentManagementController.java
index 8034966567..7ebe26611d 100644
---
a/modules/rest/src/main/java/org/apache/ignite/internal/rest/deployment/DeploymentManagementController.java
+++
b/modules/rest/src/main/java/org/apache/ignite/internal/rest/deployment/DeploymentManagementController.java
@@ -19,8 +19,6 @@ package org.apache.ignite.internal.rest.deployment;
import io.micronaut.http.annotation.Controller;
import io.micronaut.http.multipart.CompletedFileUpload;
-import java.io.IOException;
-import java.io.InputStream;
import java.util.Collection;
import java.util.HashMap;
import java.util.Map;
@@ -33,6 +31,7 @@ import org.apache.ignite.internal.deployunit.version.Version;
import org.apache.ignite.internal.rest.api.deployment.DeploymentCodeApi;
import org.apache.ignite.internal.rest.api.deployment.DeploymentInfo;
import org.apache.ignite.internal.rest.api.deployment.UnitStatus;
+import org.reactivestreams.Publisher;
/**
* Implementation of {@link DeploymentCodeApi}.
@@ -46,16 +45,10 @@ public class DeploymentManagementController implements
DeploymentCodeApi {
}
@Override
- public CompletableFuture<Boolean> deploy(String unitId, String
unitVersion, CompletedFileUpload unitContent) {
- try {
- DeploymentUnit deploymentUnit = toDeploymentUnit(unitContent);
- if (unitVersion == null || unitVersion.isBlank()) {
- return deployment.deployAsync(unitId, Version.LATEST,
deploymentUnit);
- }
- return deployment.deployAsync(unitId,
Version.parseVersion(unitVersion), deploymentUnit);
- } catch (IOException e) {
- return CompletableFuture.failedFuture(e);
- }
+ public CompletableFuture<Boolean> deploy(String unitId, String
unitVersion, Publisher<CompletedFileUpload> unitContent) {
+ CompletableFuture<DeploymentUnit> result = new CompletableFuture<>();
+ unitContent.subscribe(new CompletedFileUploadSubscriber(result));
+ return result.thenCompose(deploymentUnit ->
deployment.deployAsync(unitId, parseVersion(unitVersion), deploymentUnit));
}
@Override
@@ -92,22 +85,6 @@ public class DeploymentManagementController implements
DeploymentCodeApi {
.collect(Collectors.toList()));
}
- private static DeploymentUnit toDeploymentUnit(CompletedFileUpload
unitContent) throws IOException {
- String fileName = unitContent.getFilename();
- InputStream is = unitContent.getInputStream();
- return new DeploymentUnit() {
- @Override
- public String name() {
- return fileName;
- }
-
- @Override
- public InputStream content() {
- return is;
- }
- };
- }
-
/**
* Mapper method.
*
@@ -123,4 +100,11 @@ public class DeploymentManagementController implements
DeploymentCodeApi {
}
return new UnitStatus(status.id(), versionToDeploymentStatus);
}
+
+ private static Version parseVersion(String unitVersion) {
+ if (unitVersion == null || unitVersion.isBlank()) {
+ return Version.LATEST;
+ }
+ return Version.parseVersion(unitVersion);
+ }
}
diff --git
a/modules/runner/src/integrationTest/java/org/apache/ignite/internal/deployment/ItDeploymentUnitTest.java
b/modules/runner/src/integrationTest/java/org/apache/ignite/internal/deployment/ItDeploymentUnitTest.java
index 59634a7ea1..7f53b5e4d0 100644
---
a/modules/runner/src/integrationTest/java/org/apache/ignite/internal/deployment/ItDeploymentUnitTest.java
+++
b/modules/runner/src/integrationTest/java/org/apache/ignite/internal/deployment/ItDeploymentUnitTest.java
@@ -30,7 +30,6 @@ import static org.awaitility.Awaitility.await;
import static org.hamcrest.MatcherAssert.assertThat;
import static org.junit.jupiter.api.Assertions.assertTrue;
-import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.nio.ByteBuffer;
@@ -38,9 +37,12 @@ import java.nio.channels.SeekableByteChannel;
import java.nio.file.Files;
import java.nio.file.Path;
import java.util.Collections;
+import java.util.HashMap;
import java.util.List;
+import java.util.Map;
import java.util.Objects;
import java.util.concurrent.CompletableFuture;
+import java.util.stream.Collectors;
import org.apache.ignite.internal.ClusterPerTestIntegrationTest;
import org.apache.ignite.internal.app.IgniteImpl;
import org.apache.ignite.internal.deployunit.DeploymentInfo;
@@ -72,16 +74,20 @@ public class ItDeploymentUnitTest extends
ClusterPerTestIntegrationTest {
private DeployFile bigFile;
+ private List<DeployFile> allFiles;
+
@BeforeEach
public void generateDummy() throws IOException {
smallFile = create("small.txt", SMALL_IN_BYTES, BASE_REPLICA_TIMEOUT);
mediumFile = create("medium.txt", MEDIUM_IN_BYTES,
BASE_REPLICA_TIMEOUT * 2);
- bigFile = create("big.txt", BIG_IN_BYTES, BASE_REPLICA_TIMEOUT * 3);
+ // TODO https://issues.apache.org/jira/browse/IGNITE-19009
+ // bigFile = create("big.txt", BIG_IN_BYTES, BASE_REPLICA_TIMEOUT * 3);
+ allFiles = List.of(smallFile, mediumFile);
}
private DeployFile create(String name, long size, int replicaTimeout)
throws IOException {
DeployFile deployFile = new DeployFile(workDir.resolve(name), size,
replicaTimeout);
- deployFile.ensureExist();
+ deployFile.ensureExists();
return deployFile;
}
@@ -100,6 +106,21 @@ public class ItDeploymentUnitTest extends
ClusterPerTestIntegrationTest {
.until(() -> node(2).deployment().unitsAsync(),
willBe(Collections.singletonList(status)));
}
+ @Test
+ public void deployDirectory() {
+ String id = "test";
+ Unit unit = deployAndVerify(id, Version.parseVersion("1.1.0"), false,
allFiles, 1);
+
+ IgniteImpl cmg = cluster.node(0);
+ waitUnitReplica(cmg, unit);
+
+ UnitStatus status = buildStatus(id, unit);
+
+ await().timeout(2, SECONDS)
+ .pollDelay(500, MILLISECONDS)
+ .until(() -> node(2).deployment().unitsAsync(),
willBe(Collections.singletonList(status)));
+ }
+
@Test
public void testDeployUndeploy() {
Unit unit = deployAndVerifySmall("test",
Version.parseVersion("1.1.0"), 1);
@@ -261,22 +282,35 @@ public class ItDeploymentUnitTest extends
ClusterPerTestIntegrationTest {
}
private Unit deployAndVerify(String id, Version version, boolean force,
DeployFile file, int nodeIndex) {
+ return deployAndVerify(id, version, force, List.of(file), nodeIndex);
+ }
+
+ private Unit deployAndVerify(String id, Version version, boolean force,
List<DeployFile> files, int nodeIndex) {
IgniteImpl entryNode = node(nodeIndex);
+ List<Path> paths = files.stream()
+ .map(deployFile -> deployFile.file)
+ .collect(Collectors.toList());
+
CompletableFuture<Boolean> deploy = entryNode.deployment()
- .deployAsync(id, version, force, fromPath(file.file));
+ .deployAsync(id, version, force, fromPaths(paths));
assertThat(deploy, willBe(true));
- Unit unit = new Unit(entryNode, id, version, file);
- Path nodeUnitFile = getNodeUnitFile(unit);
- assertTrue(Files.exists(nodeUnitFile));
+ Unit unit = new Unit(entryNode, id, version, files);
+
+ Path nodeUnitDirectory = getNodeUnitDirectory(entryNode, id, version);
+
+ for (DeployFile file : files) {
+ Path filePath = nodeUnitDirectory.resolve(file.file.getFileName());
+ assertTrue(Files.exists(filePath));
+ }
return unit;
}
private Unit deployAndVerifySmall(String id, Version version, int
nodeIndex) {
- return deployAndVerifyMedium(id, version, nodeIndex);
+ return deployAndVerify(id, version, smallFile, nodeIndex);
}
private Unit deployAndVerifyMedium(String id, Version version, int
nodeIndex) {
@@ -287,35 +321,52 @@ public class ItDeploymentUnitTest extends
ClusterPerTestIntegrationTest {
return deployAndVerify(id, version, bigFile, nodeIndex);
}
- private Path getNodeUnitFile(Unit unit) {
- return getNodeUnitFile(unit.deployedNode, unit.id, unit.version,
unit.file);
- }
-
- private Path getNodeUnitFile(IgniteImpl node, String unitId, Version
unitVersion, DeployFile file) {
+ private Path getNodeUnitDirectory(IgniteImpl node, String unitId, Version
unitVersion) {
String deploymentFolder = node.nodeConfiguration()
.getConfiguration(DeploymentConfiguration.KEY)
.deploymentLocation().value();
Path resolve = workDir.resolve(node.name()).resolve(deploymentFolder);
return resolve.resolve(unitId)
- .resolve(unitVersion.render())
- .resolve(file.file.getFileName());
+ .resolve(unitVersion.render());
}
private void waitUnitReplica(IgniteImpl ignite, Unit unit) {
- Path unitPath = getNodeUnitFile(ignite, unit.id, unit.version,
unit.file);
+ Path unitDirectory = getNodeUnitDirectory(ignite, unit.id,
unit.version);
+
+ int combinedTimeout = unit.files.stream().map(file ->
file.replicaTimeout).reduce(Integer::sum).get();
- await().timeout(unit.file.replicaTimeout, SECONDS)
+ await().timeout(combinedTimeout, SECONDS)
.pollDelay(1, SECONDS)
.ignoreException(IOException.class)
- .until(() -> Files.exists(unitPath) && Files.size(unitPath) ==
unit.file.expectedSize);
+ .until(() -> {
+ for (DeployFile file : unit.files) {
+ Path filePath =
unitDirectory.resolve(file.file.getFileName());
+ if (Files.notExists(filePath) || Files.size(filePath)
!= file.expectedSize) {
+ return false;
+ }
+ }
+
+ return true;
+ });
}
private void waitUnitClean(IgniteImpl ignite, Unit unit) {
- Path unitPath = getNodeUnitFile(ignite, unit.id, unit.version,
unit.file);
+ Path unitDirectory = getNodeUnitDirectory(ignite, unit.id,
unit.version);
- await().timeout(unit.file.replicaTimeout, SECONDS)
+ int combinedTimeout = unit.files.stream().map(file ->
file.replicaTimeout).reduce(Integer::sum).get();
+
+ await().timeout(combinedTimeout, SECONDS)
.pollDelay(2, SECONDS)
- .until(() -> !Files.exists(unitPath));
+ .until(() -> {
+ for (DeployFile file : unit.files) {
+ Path filePath =
unitDirectory.resolve(file.file.getFileName());
+ if (Files.exists(filePath)) {
+ return false;
+ }
+ }
+
+ return true;
+ });
}
class Unit {
@@ -325,13 +376,13 @@ public class ItDeploymentUnitTest extends
ClusterPerTestIntegrationTest {
private final Version version;
- private final DeployFile file;
+ private final List<DeployFile> files;
- Unit(IgniteImpl deployedNode, String id, Version version, DeployFile
file) {
+ Unit(IgniteImpl deployedNode, String id, Version version,
List<DeployFile> files) {
this.deployedNode = deployedNode;
this.id = id;
this.version = version;
- this.file = file;
+ this.files = files;
}
CompletableFuture<Void> undeployAsync() {
@@ -357,7 +408,7 @@ public class ItDeploymentUnitTest extends
ClusterPerTestIntegrationTest {
this.replicaTimeout = replicaTimeout;
}
- public void ensureExist() throws IOException {
+ public void ensureExists() throws IOException {
ensureFile(file, expectedSize);
}
@@ -374,23 +425,16 @@ public class ItDeploymentUnitTest extends
ClusterPerTestIntegrationTest {
}
}
- private static DeploymentUnit fromPath(Path path) {
- Objects.requireNonNull(path);
- return new DeploymentUnit() {
-
- @Override
- public String name() {
- return path.getFileName().toString();
+ private static DeploymentUnit fromPaths(List<Path> paths) {
+ Objects.requireNonNull(paths);
+ Map<String, InputStream> map = new HashMap<>();
+ try {
+ for (Path path : paths) {
+ map.put(path.getFileName().toString(),
Files.newInputStream(path));
}
-
- @Override
- public InputStream content() {
- try {
- return new FileInputStream(path.toFile());
- } catch (IOException e) {
- throw new RuntimeException(e);
- }
- }
- };
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ return () -> map;
}
}
