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 4d09fd1644 IGNITE-18729 Add code deployment API (#1698)
4d09fd1644 is described below
commit 4d09fd164455b875ca03e86c93a5a06a0bf2e084
Author: Mikhail <[email protected]>
AuthorDate: Wed Mar 1 12:16:23 2023 +0300
IGNITE-18729 Add code deployment API (#1698)
Co-authored-by: Mikhail Pochatkin <[email protected]>
Co-authored-by: Pavel Tupitsyn <[email protected]>
---
.../src/main/java/org/apache/ignite/Ignite.java | 8 +
.../apache/ignite/deployment/DeploymentUnit.java | 39 ++
.../apache/ignite/deployment/IgniteDeployment.java | 96 +++++
.../org/apache/ignite/deployment/UnitStatus.java | 159 ++++++++
.../ignite/deployment/version/UnitVersion.java | 127 ++++++
.../apache/ignite/deployment/version/Version.java | 65 +++
.../deployment/version/VersionParseException.java | 48 +++
.../ignite/deployment/version/VersionUnitTest.java | 48 +++
.../repl/executor/ItIgnitePicocliCommandsTest.java | 4 +-
.../ignite/internal/client/TcpIgniteClient.java | 7 +
.../org/apache/ignite/client/fakes/FakeIgnite.java | 6 +
.../management/ClusterManagementGroupManager.java | 19 +
modules/code-deployment/build.gradle | 42 ++
.../internal/deployunit/DeploymentManagerImpl.java | 437 +++++++++++++++++++++
.../ignite/internal/deployunit/UnitMeta.java | 143 +++++++
.../internal/deployunit/UnitMetaSerializer.java | 93 +++++
.../DeploymentConfigurationModule.java | 41 ++
.../DeploymentConfigurationSchema.java | 34 ++
.../DeploymentUnitAlreadyExistsException.java | 48 +++
.../exception/DeploymentUnitNotFoundException.java | 35 ++
.../exception/DeploymentUnitReadException.java | 42 ++
.../deployunit/message/DeployUnitMessageTypes.java | 47 +++
.../deployunit/message/DeployUnitRequest.java | 57 +++
.../deployunit/message/DeployUnitResponse.java | 36 ++
.../deployunit/message/UndeployUnitRequest.java | 41 ++
.../deployunit/message/UndeployUnitResponse.java | 36 ++
.../ignite/deployment/UnitMetaSerializerTest.java | 82 ++++
.../apache/ignite/internal/util/IgniteUtils.java | 49 ++-
.../java/org/apache/ignite/lang/ErrorGroups.java | 25 ++
modules/runner/build.gradle | 2 +
.../internal/deployment/ItDeploymentUnitTest.java | 216 ++++++++++
.../org/apache/ignite/internal/app/IgniteImpl.java | 20 +-
settings.gradle | 2 +
33 files changed, 2131 insertions(+), 23 deletions(-)
diff --git a/modules/api/src/main/java/org/apache/ignite/Ignite.java
b/modules/api/src/main/java/org/apache/ignite/Ignite.java
index f01f8779da..4d36bbddaf 100644
--- a/modules/api/src/main/java/org/apache/ignite/Ignite.java
+++ b/modules/api/src/main/java/org/apache/ignite/Ignite.java
@@ -21,6 +21,7 @@ import java.util.Collection;
import java.util.concurrent.CompletableFuture;
import org.apache.ignite.compute.ComputeJob;
import org.apache.ignite.compute.IgniteCompute;
+import org.apache.ignite.deployment.IgniteDeployment;
import org.apache.ignite.network.ClusterNode;
import org.apache.ignite.sql.IgniteSql;
import org.apache.ignite.table.manager.IgniteTables;
@@ -67,6 +68,13 @@ public interface Ignite extends AutoCloseable {
*/
IgniteCompute compute();
+ /**
+ * Returns {@link IgniteDeployment} which can be used to deploy units.
+ *
+ * @return Deployment management object.
+ */
+ IgniteDeployment deployment();
+
/**
* Gets the cluster nodes.
* NOTE: Temporary API to enable Compute until we have proper Cluster API.
diff --git
a/modules/api/src/main/java/org/apache/ignite/deployment/DeploymentUnit.java
b/modules/api/src/main/java/org/apache/ignite/deployment/DeploymentUnit.java
new file mode 100644
index 0000000000..75660a9121
--- /dev/null
+++ b/modules/api/src/main/java/org/apache/ignite/deployment/DeploymentUnit.java
@@ -0,0 +1,39 @@
+/*
+ * 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.deployment;
+
+import java.io.InputStream;
+
+/**
+ * Deployment unit interface.
+ */
+public interface DeploymentUnit {
+ /**
+ * Unit name.
+ *
+ * @return Name of deployment unit.
+ */
+ String name();
+
+ /**
+ * Input stream with deployment unit content.
+ *
+ * @return input stream with deployment unit content.
+ */
+ InputStream content();
+}
diff --git
a/modules/api/src/main/java/org/apache/ignite/deployment/IgniteDeployment.java
b/modules/api/src/main/java/org/apache/ignite/deployment/IgniteDeployment.java
new file mode 100644
index 0000000000..5d8e17eab2
--- /dev/null
+++
b/modules/api/src/main/java/org/apache/ignite/deployment/IgniteDeployment.java
@@ -0,0 +1,96 @@
+/*
+ * 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.deployment;
+
+import java.util.List;
+import java.util.concurrent.CompletableFuture;
+import org.apache.ignite.deployment.version.Version;
+
+/**
+ * Provides access to the Deployment Unit functionality.
+ */
+public interface IgniteDeployment {
+ /**
+ * Deploy provided unit to current node with latest version.
+ *
+ * @param id Unit identifier. Not empty and not null.
+ * @param deploymentUnit Unit content.
+ * @return Future with success or not result.
+ */
+ default CompletableFuture<Boolean> deployAsync(String id, DeploymentUnit
deploymentUnit) {
+ return deployAsync(id, Version.LATEST, deploymentUnit);
+ }
+
+ /**
+ * Deploy provided unit to current node.
+ * After deploy finished, this deployment unit will be place to CMG group
asynchronously.
+ *
+ * @param id Unit identifier. Not empty and not null.
+ * @param version Unit version.
+ * @param deploymentUnit Unit content.
+ * @return Future with success or not result.
+ */
+ CompletableFuture<Boolean> deployAsync(String id, Version version,
DeploymentUnit deploymentUnit);
+
+ /**
+ * Undeploy latest version of unit with corresponding identifier.
+ *
+ * @param id Unit identifier. Not empty and not null.
+ * @return Future completed when unit will be undeployed.
+ * In case when specified unit not exist future will be failed.
+ */
+ default CompletableFuture<Void> undeployAsync(String id) {
+ return undeployAsync(id, Version.LATEST);
+ }
+
+ /**
+ * Undeploy unit with corresponding identifier and version.
+ * Note that unit files will be deleted asynchronously.
+ *
+ * @param id Unit identifier.
+ * @param version Unit version.
+ * @return Future completed when unit will be undeployed.
+ * In case when specified unit not exist future will be failed.
+ */
+ CompletableFuture<Void> undeployAsync(String id, Version version);
+
+ /**
+ * Lists all deployed units.
+ *
+ * @return Future with result.
+ */
+ CompletableFuture<List<UnitStatus>> unitsAsync();
+
+ /**
+ * List all deployed versions of the specified unit.
+ *
+ * @param id Unit identifier. Not empty and not null.
+ * @return Future with list of all available version of unit.
+ * In case when unit with specified identifier not exist future list
will be empty.
+ */
+ CompletableFuture<List<Version>> versionsAsync(String id);
+
+ /**
+ * Return status of unit with provided identifier.
+ *
+ * @param id Unit identifier. Not empty and not null.
+ * @return Future with unit status.
+ * Future will be failed if unit with specified identifier not exist.
+ */
+ CompletableFuture<UnitStatus> statusAsync(String id);
+}
diff --git
a/modules/api/src/main/java/org/apache/ignite/deployment/UnitStatus.java
b/modules/api/src/main/java/org/apache/ignite/deployment/UnitStatus.java
new file mode 100644
index 0000000000..771a0fd8db
--- /dev/null
+++ b/modules/api/src/main/java/org/apache/ignite/deployment/UnitStatus.java
@@ -0,0 +1,159 @@
+/*
+ * 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.deployment;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+import org.apache.ignite.deployment.version.Version;
+import org.apache.ignite.internal.tostring.S;
+
+/**
+ * Deployment unit status.
+ */
+public class UnitStatus {
+ /**
+ * Unit identifier.
+ */
+ private final String id;
+
+ /**
+ * Map from existing unit version to list of nodes consistent ids where
unit deployed.
+ */
+ private final Map<Version, List<String>> versionToConsistentIds;
+
+ /**
+ * Constructor.
+ *
+ * @param id Unit identifier.
+ * @param versionToConsistentIds Map from existing unit version to list
+ * of nodes consistent ids where unit deployed.
+ */
+ private UnitStatus(String id, Map<Version, List<String>>
versionToConsistentIds) {
+ this.id = id;
+ this.versionToConsistentIds =
Collections.unmodifiableMap(versionToConsistentIds);
+ }
+
+ /**
+ * Returns unit identifier.
+ *
+ * @return unit identifier.
+ */
+ public String id() {
+ return id;
+ }
+
+ /**
+ * Returns unit version.
+ *
+ * @return unit version.
+ */
+ public Set<Version> versions() {
+ return Collections.unmodifiableSet(versionToConsistentIds.keySet());
+ }
+
+ /**
+ * Returns consistent ids of nodes for provided version.
+ *
+ * @param version Unit version.
+ * @return consistent ids of nodes for provided version.
+ */
+ public List<String> consistentIds(Version version) {
+ return
Collections.unmodifiableList(versionToConsistentIds.get(version));
+ }
+
+ /**
+ * Builder provider.
+ *
+ * @param id Identifier of unit. Not null and not blank.
+ * @return Instance of {@link UnitStatusBuilder}.
+ */
+ public static UnitStatusBuilder builder(String id) {
+ Objects.requireNonNull(id);
+
+ return new UnitStatusBuilder(id);
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+ UnitStatus that = (UnitStatus) o;
+ return Objects.equals(id, that.id) &&
Objects.equals(versionToConsistentIds, that.versionToConsistentIds);
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ public int hashCode() {
+ return Objects.hash(id, versionToConsistentIds);
+ }
+
+ /** {@inheritDoc} */
+ @Override
+ public String toString() {
+ return S.toString(this);
+ }
+
+ /**
+ * Builder for {@link UnitStatus}.
+ */
+ public static class UnitStatusBuilder {
+
+ private final String id;
+ private final Map<Version, List<String>> versionToConsistentIds = new
HashMap<>();
+
+ /**
+ * Constructor.
+ *
+ * @param id unit identifier.
+ */
+ public UnitStatusBuilder(String id) {
+ this.id = id;
+ }
+
+ /**
+ * Append node consistent ids with provided version.
+ *
+ * @param version Unit version.
+ * @param consistentIds Node consistent ids.
+ * @return {@code this} builder for use in a chained invocation.
+ */
+ public UnitStatusBuilder append(Version version, List<String>
consistentIds) {
+ versionToConsistentIds.put(version, consistentIds);
+ return this;
+ }
+
+ /**
+ * Builder status method.
+ *
+ * @return {@link UnitStatus} instance.
+ */
+ public UnitStatus build() {
+ return new UnitStatus(id, versionToConsistentIds);
+ }
+ }
+}
+
diff --git
a/modules/api/src/main/java/org/apache/ignite/deployment/version/UnitVersion.java
b/modules/api/src/main/java/org/apache/ignite/deployment/version/UnitVersion.java
new file mode 100644
index 0000000000..02b2df683d
--- /dev/null
+++
b/modules/api/src/main/java/org/apache/ignite/deployment/version/UnitVersion.java
@@ -0,0 +1,127 @@
+/*
+ * 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.deployment.version;
+
+import java.util.Objects;
+
+/**
+ * Implementation of {@link Version} interface based on the three numbers
format,
+ * like x.x.x. where x is short number.
+ */
+public class UnitVersion implements Version {
+ private final short major;
+ private final short minor;
+ private final short patch;
+
+ /**
+ * Constructor.
+ *
+ * @param major Major part of version.
+ * @param minor Minor part of version.
+ * @param patch Patch part of version.
+ */
+ public UnitVersion(short major, short minor, short patch) {
+ this.major = major;
+ this.minor = minor;
+ this.patch = patch;
+ }
+
+ @Override
+ public String render() {
+ return major + "." + minor + "." + patch;
+ }
+
+ /**
+ * Parse string representation of version to {@link UnitVersion} if
possible.
+ *
+ * @param s String representation of version.
+ * @return Instance of {@link UnitVersion}.
+ * @throws VersionParseException in case when string is not required
{@link UnitVersion} format.
+ */
+ public static UnitVersion parse(String s) {
+ Objects.requireNonNull(s);
+ try {
+ String[] split = s.split("\\.", -1);
+ if (split.length > 3 || split.length == 0) {
+ throw new VersionParseException("Invalid version format");
+ }
+
+ short major = Short.parseShort(split[0]);
+ short minor = split.length > 1 ? Short.parseShort(split[1]) : 0;
+ short patch = split.length > 2 ? Short.parseShort(split[2]) : 0;
+
+ return new UnitVersion(major, minor, patch);
+ } catch (NumberFormatException e) {
+ throw new VersionParseException(e);
+ }
+ }
+
+ @Override
+ public int compareTo(Version o) {
+ if (o == LATEST) {
+ return -1;
+ }
+
+ UnitVersion version = (UnitVersion) o;
+
+ int majorCompare = Short.compare(major, version.major);
+ if (majorCompare != 0) {
+ return majorCompare;
+ }
+
+ int minorCompare = Short.compare(minor, version.minor);
+ if (minorCompare != 0) {
+ return minorCompare;
+ }
+
+ return Short.compare(patch, version.patch);
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+
+ UnitVersion version = (UnitVersion) o;
+
+ if (major != version.major) {
+ return false;
+ }
+ if (minor != version.minor) {
+ return false;
+ }
+ return patch == version.patch;
+ }
+
+ @Override
+ public int hashCode() {
+ int result = major;
+ result = 31 * result + minor;
+ result = 31 * result + patch;
+ return result;
+ }
+
+ @Override
+ public String toString() {
+ return render();
+ }
+}
diff --git
a/modules/api/src/main/java/org/apache/ignite/deployment/version/Version.java
b/modules/api/src/main/java/org/apache/ignite/deployment/version/Version.java
new file mode 100644
index 0000000000..f9a4e29847
--- /dev/null
+++
b/modules/api/src/main/java/org/apache/ignite/deployment/version/Version.java
@@ -0,0 +1,65 @@
+/*
+ * 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.deployment.version;
+
+
+/**
+ * Unit version interface. Version implementations should be comparable.
+ */
+public interface Version extends Comparable<Version> {
+ /**
+ * Render version representation in String format.
+ *
+ * @return version string representation.
+ */
+ String render();
+
+ /**
+ * Implementation of {@link Version} interface with special latest logic.
+ * This version have special unique representation. Moreover by convention
+ * this implementation should be oldest version of any another independent
of implementation.
+ */
+ Version LATEST = new Version() {
+ @Override
+ public String render() {
+ return "latest";
+ }
+
+ @Override
+ public int compareTo(Version o) {
+ if (o == LATEST) {
+ return 0;
+ }
+ return 1;
+ }
+ };
+
+ /**
+ * Parse version from String.
+ *
+ * @param s string version representation.
+ * @return Version instance of
+ */
+ static Version parseVersion(String s) {
+ if ("latest".equals(s)) {
+ return LATEST;
+ }
+
+ return UnitVersion.parse(s);
+ }
+}
diff --git
a/modules/api/src/main/java/org/apache/ignite/deployment/version/VersionParseException.java
b/modules/api/src/main/java/org/apache/ignite/deployment/version/VersionParseException.java
new file mode 100644
index 0000000000..717035a9c0
--- /dev/null
+++
b/modules/api/src/main/java/org/apache/ignite/deployment/version/VersionParseException.java
@@ -0,0 +1,48 @@
+/*
+ * 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.deployment.version;
+
+/**
+ * Throws when {@link Version} of deployment unit not parsable.
+ */
+public class VersionParseException extends RuntimeException {
+ /**
+ * Constructor.
+ */
+ public VersionParseException() {
+
+ }
+
+ /**
+ * Constructor.
+ *
+ * @param cause Cause error.
+ */
+ public VersionParseException(Throwable cause) {
+ super(cause);
+ }
+
+ /**
+ * Constructor.
+ *
+ * @param message Error message.
+ */
+ public VersionParseException(String message) {
+ super(message);
+ }
+}
diff --git
a/modules/api/src/test/java/org/apache/ignite/deployment/version/VersionUnitTest.java
b/modules/api/src/test/java/org/apache/ignite/deployment/version/VersionUnitTest.java
new file mode 100644
index 0000000000..bd0ed3cf44
--- /dev/null
+++
b/modules/api/src/test/java/org/apache/ignite/deployment/version/VersionUnitTest.java
@@ -0,0 +1,48 @@
+/*
+ * 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.deployment.version;
+
+import static org.hamcrest.MatcherAssert.assertThat;
+import static org.hamcrest.Matchers.is;
+
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Unit tests for {@link Version}.
+ */
+public class VersionUnitTest {
+ @Test
+ public void testParseCorrect() {
+ assertThat(Version.parseVersion("1.1.0"), is(new UnitVersion((short)
1, (short) 1, (short) 0)));
+ assertThat(Version.parseVersion("1.1.01"), is(new UnitVersion((short)
1, (short) 1, (short) 1)));
+ assertThat(Version.parseVersion("1.1"), is(new UnitVersion((short) 1,
(short) 1, (short) 0)));
+ assertThat(Version.parseVersion("1"), is(new UnitVersion((short) 1,
(short) 0, (short) 0)));
+ assertThat(Version.parseVersion("latest"), is(Version.LATEST));
+ }
+
+ @Test
+ public void testParseErrors() {
+ Assertions.assertThrows(VersionParseException.class, () ->
Version.parseVersion("1.1.1.1"));
+ Assertions.assertThrows(VersionParseException.class, () ->
Version.parseVersion(""));
+ Assertions.assertThrows(VersionParseException.class, () ->
Version.parseVersion("version"));
+ Assertions.assertThrows(VersionParseException.class, () ->
Version.parseVersion("1."));
+ Assertions.assertThrows(VersionParseException.class, () ->
Version.parseVersion(String.valueOf(Integer.MAX_VALUE)));
+ Assertions.assertThrows(VersionParseException.class, () ->
Version.parseVersion("1.1f"));
+ }
+}
diff --git
a/modules/cli/src/integrationTest/java/org/apache/ignite/internal/cli/core/repl/executor/ItIgnitePicocliCommandsTest.java
b/modules/cli/src/integrationTest/java/org/apache/ignite/internal/cli/core/repl/executor/ItIgnitePicocliCommandsTest.java
index 122a9befcc..b4b8a43421 100644
---
a/modules/cli/src/integrationTest/java/org/apache/ignite/internal/cli/core/repl/executor/ItIgnitePicocliCommandsTest.java
+++
b/modules/cli/src/integrationTest/java/org/apache/ignite/internal/cli/core/repl/executor/ItIgnitePicocliCommandsTest.java
@@ -216,7 +216,7 @@ public class ItIgnitePicocliCommandsTest extends
CliCommandTestInitializedIntegr
// wait for lazy init of node config completer
await("For given parsed words: " + givenParsedLine.words()).until(
() -> complete(givenParsedLine),
- containsInAnyOrder("rest", "compute", "clientConnector",
"raft", "network", "cluster")
+ containsInAnyOrder("rest", "compute", "clientConnector",
"raft", "network", "cluster", "deployment")
);
}
@@ -243,7 +243,7 @@ public class ItIgnitePicocliCommandsTest extends
CliCommandTestInitializedIntegr
// wait for lazy init of node config completer
await("For given parsed words: " + givenParsedLine.words()).until(
() -> complete(givenParsedLine),
- containsInAnyOrder("rest", "clientConnector", "network",
"cluster")
+ containsInAnyOrder("rest", "clientConnector", "network",
"cluster", "deployment")
);
}
diff --git
a/modules/client/src/main/java/org/apache/ignite/internal/client/TcpIgniteClient.java
b/modules/client/src/main/java/org/apache/ignite/internal/client/TcpIgniteClient.java
index 8474f280ec..8ab6beaafa 100644
---
a/modules/client/src/main/java/org/apache/ignite/internal/client/TcpIgniteClient.java
+++
b/modules/client/src/main/java/org/apache/ignite/internal/client/TcpIgniteClient.java
@@ -27,6 +27,7 @@ import java.util.function.BiFunction;
import org.apache.ignite.client.IgniteClient;
import org.apache.ignite.client.IgniteClientConfiguration;
import org.apache.ignite.compute.IgniteCompute;
+import org.apache.ignite.deployment.IgniteDeployment;
import org.apache.ignite.internal.client.compute.ClientCompute;
import org.apache.ignite.internal.client.io.ClientConnectionMultiplexer;
import org.apache.ignite.internal.client.proto.ClientOp;
@@ -139,6 +140,12 @@ public class TcpIgniteClient implements IgniteClient {
return compute;
}
+ /** {@inheritDoc} */
+ @Override
+ public IgniteDeployment deployment() {
+ throw new UnsupportedOperationException("Deployment management not
implemented yet.");
+ }
+
/** {@inheritDoc} */
@Override
public Collection<ClusterNode> clusterNodes() {
diff --git
a/modules/client/src/test/java/org/apache/ignite/client/fakes/FakeIgnite.java
b/modules/client/src/test/java/org/apache/ignite/client/fakes/FakeIgnite.java
index 51da2b6737..16deb7367f 100644
---
a/modules/client/src/test/java/org/apache/ignite/client/fakes/FakeIgnite.java
+++
b/modules/client/src/test/java/org/apache/ignite/client/fakes/FakeIgnite.java
@@ -22,6 +22,7 @@ import java.util.UUID;
import java.util.concurrent.CompletableFuture;
import org.apache.ignite.Ignite;
import org.apache.ignite.compute.IgniteCompute;
+import org.apache.ignite.deployment.IgniteDeployment;
import org.apache.ignite.internal.hlc.HybridTimestamp;
import org.apache.ignite.internal.replicator.ReplicationGroupId;
import org.apache.ignite.internal.sql.engine.QueryProcessor;
@@ -167,6 +168,11 @@ public class FakeIgnite implements Ignite {
throw new UnsupportedOperationException("Not implemented yet");
}
+ @Override
+ public IgniteDeployment deployment() {
+ throw new UnsupportedOperationException("Not implemented yet");
+ }
+
/** {@inheritDoc} */
@Override
public Collection<ClusterNode> clusterNodes() {
diff --git
a/modules/cluster-management/src/main/java/org/apache/ignite/internal/cluster/management/ClusterManagementGroupManager.java
b/modules/cluster-management/src/main/java/org/apache/ignite/internal/cluster/management/ClusterManagementGroupManager.java
index f80852a62a..aa9a678cba 100644
---
a/modules/cluster-management/src/main/java/org/apache/ignite/internal/cluster/management/ClusterManagementGroupManager.java
+++
b/modules/cluster-management/src/main/java/org/apache/ignite/internal/cluster/management/ClusterManagementGroupManager.java
@@ -714,6 +714,25 @@ public class ClusterManagementGroupManager implements
IgniteComponent {
}
}
+ /**
+ * Returns a future that, when complete, resolves into a list of node
names that host the CMG.
+ *
+ * @return Future that, when complete, resolves into a list of node names
that host the CMG.
+ */
+ public CompletableFuture<Set<String>> cmgNodes() {
+ if (!busyLock.enterBusy()) {
+ return failedFuture(new NodeStoppingException());
+ }
+
+ try {
+ return raftServiceAfterJoin()
+ .thenCompose(CmgRaftService::readClusterState)
+ .thenApply(ClusterState::cmgNodes);
+ } finally {
+ busyLock.leaveBusy();
+ }
+ }
+
/**
* Returns a future that, when complete, resolves into a logical topology
snapshot.
*
diff --git a/modules/code-deployment/build.gradle
b/modules/code-deployment/build.gradle
new file mode 100644
index 0000000000..2a7af850d6
--- /dev/null
+++ b/modules/code-deployment/build.gradle
@@ -0,0 +1,42 @@
+/*
+ * 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.
+ */
+
+apply from: "$rootDir/buildscripts/java-core.gradle"
+apply from: "$rootDir/buildscripts/java-integration-test.gradle"
+apply from: "$rootDir/buildscripts/publishing.gradle"
+apply from: "$rootDir/buildscripts/java-junit5.gradle"
+
+description = 'ignite-code-deployment'
+
+dependencies {
+ annotationProcessor project(':ignite-configuration-annotation-processor')
+ annotationProcessor project(':ignite-network-annotation-processor')
+ annotationProcessor libs.auto.service
+
+ implementation libs.auto.service.annotations
+ implementation project(':ignite-core')
+ implementation project(':ignite-api')
+ implementation project(':ignite-network-api')
+ implementation project(':ignite-metastorage-api')
+ implementation project(':ignite-cluster-management')
+ implementation project(':ignite-configuration')
+ implementation project(':ignite-network')
+ implementation project(':ignite-schema')
+
+ testImplementation libs.hamcrest.core
+ testImplementation libs.hamcrest.optional
+}
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
new file mode 100644
index 0000000000..2c2869a503
--- /dev/null
+++
b/modules/code-deployment/src/main/java/org/apache/ignite/internal/deployunit/DeploymentManagerImpl.java
@@ -0,0 +1,437 @@
+/*
+ * 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.deployunit;
+
+import static java.nio.file.StandardCopyOption.ATOMIC_MOVE;
+import static java.nio.file.StandardCopyOption.REPLACE_EXISTING;
+import static java.nio.file.StandardOpenOption.CREATE;
+import static java.nio.file.StandardOpenOption.SYNC;
+import static java.nio.file.StandardOpenOption.TRUNCATE_EXISTING;
+import static org.apache.ignite.internal.metastorage.dsl.Conditions.exists;
+import static org.apache.ignite.internal.metastorage.dsl.Conditions.notExists;
+import static org.apache.ignite.internal.metastorage.dsl.Conditions.revision;
+import static org.apache.ignite.internal.metastorage.dsl.Operations.put;
+
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.ArrayList;
+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.concurrent.Flow.Subscriber;
+import java.util.concurrent.Flow.Subscription;
+import java.util.stream.Collectors;
+import org.apache.ignite.deployment.DeploymentUnit;
+import org.apache.ignite.deployment.IgniteDeployment;
+import org.apache.ignite.deployment.UnitStatus;
+import org.apache.ignite.deployment.UnitStatus.UnitStatusBuilder;
+import org.apache.ignite.deployment.version.Version;
+import
org.apache.ignite.internal.cluster.management.ClusterManagementGroupManager;
+import
org.apache.ignite.internal.deployunit.configuration.DeploymentConfiguration;
+import
org.apache.ignite.internal.deployunit.exception.DeploymentUnitAlreadyExistsException;
+import
org.apache.ignite.internal.deployunit.exception.DeploymentUnitNotFoundException;
+import
org.apache.ignite.internal.deployunit.exception.DeploymentUnitReadException;
+import org.apache.ignite.internal.deployunit.message.DeployUnitMessageTypes;
+import org.apache.ignite.internal.deployunit.message.DeployUnitRequest;
+import org.apache.ignite.internal.deployunit.message.DeployUnitRequestBuilder;
+import org.apache.ignite.internal.deployunit.message.DeployUnitRequestImpl;
+import org.apache.ignite.internal.deployunit.message.DeployUnitResponse;
+import org.apache.ignite.internal.deployunit.message.DeployUnitResponseBuilder;
+import org.apache.ignite.internal.deployunit.message.DeployUnitResponseImpl;
+import org.apache.ignite.internal.deployunit.message.UndeployUnitRequest;
+import org.apache.ignite.internal.deployunit.message.UndeployUnitRequestImpl;
+import org.apache.ignite.internal.deployunit.message.UndeployUnitResponse;
+import org.apache.ignite.internal.deployunit.message.UndeployUnitResponseImpl;
+import org.apache.ignite.internal.future.InFlightFutures;
+import org.apache.ignite.internal.logger.IgniteLogger;
+import org.apache.ignite.internal.logger.Loggers;
+import org.apache.ignite.internal.manager.IgniteComponent;
+import org.apache.ignite.internal.metastorage.Entry;
+import org.apache.ignite.internal.metastorage.MetaStorageManager;
+import org.apache.ignite.internal.metastorage.dsl.Operation;
+import org.apache.ignite.internal.metastorage.dsl.Operations;
+import org.apache.ignite.internal.util.IgniteUtils;
+import org.apache.ignite.lang.ByteArray;
+import org.apache.ignite.network.ClusterNode;
+import org.apache.ignite.network.ClusterService;
+
+//TODO: rework metastorage keys IGNITE-18870
+/**
+ * Deployment manager implementation.
+ */
+public class DeploymentManagerImpl implements IgniteDeployment,
IgniteComponent {
+
+ private static final IgniteLogger LOG =
Loggers.forClass(DeploymentManagerImpl.class);
+
+ private static final String TMP_SUFFIX = ".tmp";
+
+ private static final String DEPLOY_UNIT_PREFIX = "deploy-unit.";
+
+ private static final String UNITS_PREFIX = DEPLOY_UNIT_PREFIX + "units.";
+
+ /**
+ * Meta storage.
+ */
+ private final MetaStorageManager metaStorage;
+
+ /**
+ * Deployment configuration.
+ */
+ private final DeploymentConfiguration configuration;
+
+ /**
+ * Cluster management group manager.
+ */
+ private final ClusterManagementGroupManager cmgManager;
+
+ /**
+ * In flight futures tracker.
+ */
+ private final InFlightFutures inFlightFutures = new InFlightFutures();
+
+ /**
+ * Cluster service.
+ */
+ private final ClusterService clusterService;
+
+ /**
+ * Folder for units.
+ */
+ private Path unitsFolder;
+
+ /**
+ * Constructor.
+ *
+ * @param clusterService Cluster service.
+ * @param metaStorage Meta storage.
+ * @param workDir Node working directory.
+ * @param configuration Deployment configuration.
+ * @param cmgManager Cluster management group manager.
+ */
+ public DeploymentManagerImpl(ClusterService clusterService,
+ MetaStorageManager metaStorage,
+ Path workDir,
+ DeploymentConfiguration configuration,
+ ClusterManagementGroupManager cmgManager) {
+ this.clusterService = clusterService;
+ this.metaStorage = metaStorage;
+ this.configuration = configuration;
+ this.cmgManager = cmgManager;
+ unitsFolder = workDir;
+ }
+
+ @Override
+ public CompletableFuture<Boolean> deployAsync(String id, Version version,
DeploymentUnit deploymentUnit) {
+ checkId(id);
+ Objects.requireNonNull(version);
+ Objects.requireNonNull(deploymentUnit);
+
+ ByteArray key = new ByteArray(UNITS_PREFIX + id + ":" +
version.render());
+
+ UnitMeta meta = new UnitMeta(id, version, deploymentUnit.name(),
Collections.emptyList());
+
+ Operation put = put(key, UnitMetaSerializer.serialize(meta));
+
+ DeployUnitRequestBuilder builder = DeployUnitRequestImpl.builder();
+
+ try {
+ builder.unitContent(deploymentUnit.content().readAllBytes());
+ } catch (IOException e) {
+ LOG.error("Error to read deployment unit content", e);
+ return CompletableFuture.failedFuture(new
DeploymentUnitReadException(e));
+ }
+ DeployUnitRequest request = builder
+ .unitName(deploymentUnit.name())
+ .id(id)
+ .version(version.render())
+ .build();
+
+ return metaStorage.invoke(notExists(key), put, Operations.noop())
+ .thenCompose(success -> {
+ if (success) {
+ return doDeploy(request);
+ }
+ LOG.error("Failed to deploy meta of unit " + id + ":" +
version);
+ return CompletableFuture.failedFuture(
+ new DeploymentUnitAlreadyExistsException(id,
+ "Unit " + id + ":" + version + " already
exists"));
+ })
+ .thenApply(completed -> {
+ if (completed) {
+ startDeployAsyncToCmg(request);
+ }
+ return completed;
+ });
+ }
+
+ private void startDeployAsyncToCmg(DeployUnitRequest request) {
+ cmgManager.cmgNodes()
+ .thenAccept(nodes -> {
+ for (String node : nodes) {
+ ClusterNode clusterNode =
clusterService.topologyService().getByConsistentId(node);
+ if (clusterNode != null) {
+
inFlightFutures.registerFuture(requestDeploy(clusterNode, request));
+ }
+ }
+ });
+ }
+
+ private CompletableFuture<Boolean> requestDeploy(ClusterNode clusterNode,
DeployUnitRequest request) {
+ return clusterService.messagingService()
+ .invoke(clusterNode, request, Long.MAX_VALUE)
+ .thenCompose(message -> {
+ Throwable error = ((DeployUnitResponse) message).error();
+ if (error != null) {
+ LOG.error("Failed to deploy unit " + request.id() +
":" + request.version()
+ + " to node " + clusterNode, error);
+ return CompletableFuture.failedFuture(error);
+ }
+ return CompletableFuture.completedFuture(true);
+ });
+ }
+
+ @Override
+ public CompletableFuture<Void> undeployAsync(String id, Version version) {
+ checkId(id);
+ Objects.requireNonNull(version);
+
+ ByteArray key = new ByteArray(UNITS_PREFIX + id + ":" + version);
+
+ return metaStorage.invoke(exists(key), Operations.remove(key),
Operations.noop())
+ .thenCompose(success -> {
+ if (success) {
+ return cmgManager.logicalTopology();
+ }
+ return CompletableFuture.failedFuture(new
DeploymentUnitNotFoundException(
+ "Unit " + id + " with version " + version + "
doesn't exist"));
+ }).thenApply(logicalTopologySnapshot -> {
+ for (ClusterNode node : logicalTopologySnapshot.nodes()) {
+ clusterService.messagingService()
+ .invoke(node, UndeployUnitRequestImpl.builder()
+ .id(id)
+ .version(version.render())
+ .build(),
+ Long.MAX_VALUE)
+ .thenAccept(message -> {
+ Throwable error = ((UndeployUnitResponse)
message).error();
+ if (error != null) {
+ LOG.error("Failed to undeploy unit " +
id + ":" + version
+ + " from node " + node, error);
+ }
+ });
+ }
+ return null;
+ });
+ }
+
+ @Override
+ public CompletableFuture<List<UnitStatus>> unitsAsync() {
+ CompletableFuture<List<UnitStatus>> result = new CompletableFuture<>();
+ Map<String, UnitStatusBuilder> map = new HashMap<>();
+ metaStorage.prefix(new ByteArray(UNITS_PREFIX))
+ .subscribe(new Subscriber<>() {
+ @Override
+ public void onSubscribe(Subscription subscription) {
+ subscription.request(Long.MAX_VALUE);
+ }
+
+ @Override
+ public void onNext(Entry item) {
+ UnitMeta meta =
UnitMetaSerializer.deserialize(item.value());
+ map.computeIfAbsent(meta.id(), UnitStatus::builder)
+ .append(meta.version(),
meta.consistentIdLocation());
+ }
+
+ @Override
+ public void onError(Throwable throwable) {
+ result.completeExceptionally(throwable);
+ }
+
+ @Override
+ public void onComplete() {
+
result.complete(map.values().stream().map(UnitStatusBuilder::build).collect(Collectors.toList()));
+ }
+ });
+ return result;
+ }
+
+ @Override
+ public CompletableFuture<List<Version>> versionsAsync(String id) {
+ checkId(id);
+ CompletableFuture<List<Version>> result = new CompletableFuture<>();
+ metaStorage.prefix(new ByteArray(UNITS_PREFIX + id))
+ .subscribe(new Subscriber<>() {
+ private final List<Version> list = new ArrayList<>();
+
+ @Override
+ public void onSubscribe(Subscription subscription) {
+ subscription.request(Long.MAX_VALUE);
+ }
+
+ @Override
+ public void onNext(Entry item) {
+ UnitMeta deserialize =
UnitMetaSerializer.deserialize(item.value());
+ list.add(deserialize.version());
+ }
+
+ @Override
+ public void onError(Throwable throwable) {
+ result.completeExceptionally(throwable);
+ }
+
+ @Override
+ public void onComplete() {
+ result.complete(list);
+ }
+ });
+ return result;
+ }
+
+ @Override
+ public CompletableFuture<UnitStatus> statusAsync(String id) {
+ checkId(id);
+ CompletableFuture<UnitStatus> result = new CompletableFuture<>();
+ metaStorage.prefix(new ByteArray(UNITS_PREFIX + id))
+ .subscribe(new Subscriber<>() {
+ private UnitStatusBuilder builder;
+
+ @Override
+ public void onSubscribe(Subscription subscription) {
+ subscription.request(Long.MAX_VALUE);
+ }
+
+ @Override
+ public void onNext(Entry item) {
+ if (builder == null) {
+ builder = UnitStatus.builder(id);
+ }
+ UnitMeta deserialize =
UnitMetaSerializer.deserialize(item.value());
+ builder.append(deserialize.version(),
deserialize.consistentIdLocation());
+ }
+
+ @Override
+ public void onError(Throwable throwable) {
+ result.completeExceptionally(throwable);
+ }
+
+ @Override
+ public void onComplete() {
+ if (builder != null) {
+ result.complete(builder.build());
+ } else {
+ result.completeExceptionally(
+ new DeploymentUnitNotFoundException("Unit
with " + id + " doesn't exist."));
+ }
+ }
+ });
+ return result;
+ }
+
+ @Override
+ public void start() {
+ unitsFolder =
unitsFolder.resolve(configuration.deploymentLocation().value());
+
clusterService.messagingService().addMessageHandler(DeployUnitMessageTypes.class,
+ (message, senderConsistentId, correlationId) -> {
+ if (message instanceof DeployUnitRequest) {
+ processDeployRequest((DeployUnitRequest) message,
senderConsistentId, correlationId);
+ } else if (message instanceof UndeployUnitRequest) {
+ processUndeployRequest((UndeployUnitRequest) message,
senderConsistentId, correlationId);
+ }
+ });
+ }
+
+ private void processDeployRequest(DeployUnitRequest executeRequest, String
senderConsistentId, long correlationId) {
+ doDeploy(executeRequest).whenComplete((success, throwable) -> {
+ DeployUnitResponseBuilder builder =
DeployUnitResponseImpl.builder();
+ if (throwable != null) {
+ builder.error(throwable);
+ }
+ clusterService.messagingService().respond(senderConsistentId,
+ builder.build(), correlationId);
+ });
+ }
+
+ private void processUndeployRequest(UndeployUnitRequest executeRequest,
String senderConsistentId, long correlationId) {
+ try {
+ Path unitPath = unitsFolder
+ .resolve(executeRequest.id())
+ .resolve(executeRequest.version());
+
+ IgniteUtils.deleteIfExistsThrowable(unitPath);
+ } catch (IOException e) {
+ LOG.error("Failed to undeploy unit " + executeRequest.id() + ":" +
executeRequest.version(), e);
+ clusterService.messagingService()
+ .respond(senderConsistentId,
UndeployUnitResponseImpl.builder().error(e).build(), correlationId);
+ return;
+ }
+
+ clusterService.messagingService()
+ .respond(senderConsistentId,
UndeployUnitResponseImpl.builder().build(), correlationId);
+ }
+
+ private CompletableFuture<Boolean> doDeploy(DeployUnitRequest
executeRequest) {
+ String id = executeRequest.id();
+ String version = executeRequest.version();
+ try {
+ Path unitPath = unitsFolder
+ .resolve(executeRequest.id())
+ .resolve(executeRequest.version())
+ .resolve(executeRequest.unitName());
+
+ Path unitPathTmp = unitPath.resolveSibling(unitPath.getFileName()
+ TMP_SUFFIX);
+
+ Files.createDirectories(unitPathTmp.getParent());
+
+ Files.write(unitPathTmp, executeRequest.unitContent(), CREATE,
SYNC, TRUNCATE_EXISTING);
+ Files.move(unitPathTmp, unitPath, ATOMIC_MOVE, REPLACE_EXISTING);
+ } catch (IOException e) {
+ LOG.error("Failed to deploy unit " + executeRequest.id() + ":" +
executeRequest.version(), e);
+ return CompletableFuture.failedFuture(e);
+ }
+
+ ByteArray key = new ByteArray(UNITS_PREFIX + id + ":" + version);
+ return metaStorage.get(key)
+ .thenCompose(e -> {
+ UnitMeta prev = UnitMetaSerializer.deserialize(e.value());
+
+
prev.addConsistentId(clusterService.topologyService().localMember().name());
+
+ return metaStorage.invoke(
+ revision(key).eq(e.revision()),
+ put(key, UnitMetaSerializer.serialize(prev)),
+ Operations.noop());
+ });
+ }
+
+ @Override
+ public void stop() throws Exception {
+ inFlightFutures.cancelInFlightFutures();
+ }
+
+ private static void checkId(String id) {
+ Objects.requireNonNull(id);
+
+ if (id.isBlank()) {
+ throw new IllegalArgumentException("Id is blank");
+ }
+ }
+}
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
new file mode 100644
index 0000000000..5730f8cc7a
--- /dev/null
+++
b/modules/code-deployment/src/main/java/org/apache/ignite/internal/deployunit/UnitMeta.java
@@ -0,0 +1,143 @@
+/*
+ * 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.deployunit;
+
+import java.util.ArrayList;
+import java.util.List;
+import org.apache.ignite.deployment.version.Version;
+
+/**
+ * Unit meta data class.
+ */
+public class UnitMeta {
+ /**
+ * Unit id.
+ */
+ private final String id;
+
+ /**
+ * Unit version.
+ */
+ private final Version version;
+
+ /**
+ * Unit name.
+ */
+ private final String name;
+
+ /**
+ * Consistent ids of nodes with.
+ */
+ private final List<String> consistentIdLocation = new ArrayList<>();
+
+ /**
+ * Constructor.
+ *
+ * @param id Unit identifier.
+ * @param version Unit version.
+ * @param name Unit name.
+ * @param consistentIdLocation Consistent ids of nodes where unit deployed.
+ */
+ public UnitMeta(String id, Version version, String name, List<String>
consistentIdLocation) {
+ this.id = id;
+ this.version = version;
+ this.name = name;
+ this.consistentIdLocation.addAll(consistentIdLocation);
+ }
+
+ public UnitMeta(String id, String name, List<String> consistentIdLocation)
{
+ this(id, Version.LATEST, name, consistentIdLocation);
+ }
+
+ /**
+ * Returns identifier of deployment unit.
+ *
+ * @return Identifier of deployment unit.
+ */
+ public String id() {
+ return id;
+ }
+
+ /**
+ * Returns version of deployment unit.
+ *
+ * @return Version of deployment unit.
+ */
+ public Version version() {
+ return version;
+ }
+
+ /**
+ * Returns name of deployment unit.
+ *
+ * @return name of deployment unit.
+ */
+ public String name() {
+ return name;
+ }
+
+ /**
+ * Returns list of nodes consistent id where deployment unit deployed.
+ *
+ * @return List of nodes consistent id where deployment unit deployed.
+ */
+ public List<String> consistentIdLocation() {
+ return consistentIdLocation;
+ }
+
+ /**
+ * Register node as deployment unit holder.
+ *
+ * @param id Consistent identifier of node.
+ */
+ public void addConsistentId(String id) {
+ consistentIdLocation.add(id);
+ }
+
+ @Override
+ public boolean equals(Object o) {
+ if (this == o) {
+ return true;
+ }
+ if (o == null || getClass() != o.getClass()) {
+ return false;
+ }
+
+ UnitMeta meta = (UnitMeta) o;
+
+ if (id != null ? !id.equals(meta.id) : meta.id != null) {
+ return false;
+ }
+ if (version != null ? !version.equals(meta.version) : meta.version !=
null) {
+ return false;
+ }
+ if (name != null ? !name.equals(meta.name) : meta.name != null) {
+ return false;
+ }
+ return consistentIdLocation.equals(meta.consistentIdLocation);
+ }
+
+ @Override
+ 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 + consistentIdLocation.hashCode();
+ return result;
+ }
+}
diff --git
a/modules/code-deployment/src/main/java/org/apache/ignite/internal/deployunit/UnitMetaSerializer.java
b/modules/code-deployment/src/main/java/org/apache/ignite/internal/deployunit/UnitMetaSerializer.java
new file mode 100644
index 0000000000..1101342177
--- /dev/null
+++
b/modules/code-deployment/src/main/java/org/apache/ignite/internal/deployunit/UnitMetaSerializer.java
@@ -0,0 +1,93 @@
+/*
+ * 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.deployunit;
+
+import static java.nio.charset.StandardCharsets.UTF_8;
+
+import java.util.ArrayList;
+import java.util.Base64;
+import java.util.Base64.Decoder;
+import java.util.Base64.Encoder;
+import java.util.List;
+import org.apache.ignite.deployment.version.Version;
+
+/**
+ * Serializer for {@link UnitMeta}.
+ */
+public class UnitMetaSerializer {
+ private static final String SEPARATOR = ";";
+
+ /**
+ * Constructor.
+ */
+ private UnitMetaSerializer() {
+
+ }
+
+ /**
+ * Serialize unit meta.
+ *
+ * @param meta Unit meta.
+ * @return Serialized unit meta.
+ */
+ public static byte[] serialize(UnitMeta meta) {
+ StringBuilder sb = new StringBuilder();
+
+ appendWithEncoding(sb, meta.id());
+ appendWithEncoding(sb, meta.version().render());
+ appendWithEncoding(sb, meta.name());
+
+ for (String id : meta.consistentIdLocation()) {
+ appendWithEncoding(sb, id);
+ }
+
+ 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.
+ *
+ * @param bytes Byte array.
+ * @return Unit meta.
+ */
+ public static UnitMeta deserialize(byte[] bytes) {
+ String s = new String(bytes, UTF_8);
+
+ String[] split = s.split(SEPARATOR);
+
+ Decoder decoder = Base64.getDecoder();
+
+ 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 = new ArrayList<>();
+ for (int i = 3; i < split.length; i++) {
+ ids.add(new String(decoder.decode(split[i]), UTF_8));
+ }
+
+ return new UnitMeta(id, Version.parseVersion(version), unitName, ids);
+ }
+}
diff --git
a/modules/code-deployment/src/main/java/org/apache/ignite/internal/deployunit/configuration/DeploymentConfigurationModule.java
b/modules/code-deployment/src/main/java/org/apache/ignite/internal/deployunit/configuration/DeploymentConfigurationModule.java
new file mode 100644
index 0000000000..3d0cc121cf
--- /dev/null
+++
b/modules/code-deployment/src/main/java/org/apache/ignite/internal/deployunit/configuration/DeploymentConfigurationModule.java
@@ -0,0 +1,41 @@
+/*
+ * 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.deployunit.configuration;
+
+import com.google.auto.service.AutoService;
+import java.util.Collection;
+import java.util.Collections;
+import org.apache.ignite.configuration.RootKey;
+import org.apache.ignite.configuration.annotation.ConfigurationType;
+import org.apache.ignite.internal.configuration.ConfigurationModule;
+
+/**
+ * Configuration module for {@link DeploymentConfiguration}.
+ */
+@AutoService(ConfigurationModule.class)
+public class DeploymentConfigurationModule implements ConfigurationModule {
+ @Override
+ public ConfigurationType type() {
+ return ConfigurationType.LOCAL;
+ }
+
+ @Override
+ public Collection<RootKey<?, ?>> rootKeys() {
+ return Collections.singleton(DeploymentConfiguration.KEY);
+ }
+}
diff --git
a/modules/code-deployment/src/main/java/org/apache/ignite/internal/deployunit/configuration/DeploymentConfigurationSchema.java
b/modules/code-deployment/src/main/java/org/apache/ignite/internal/deployunit/configuration/DeploymentConfigurationSchema.java
new file mode 100644
index 0000000000..a8e3a9d21f
--- /dev/null
+++
b/modules/code-deployment/src/main/java/org/apache/ignite/internal/deployunit/configuration/DeploymentConfigurationSchema.java
@@ -0,0 +1,34 @@
+/*
+ * 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.deployunit.configuration;
+
+import org.apache.ignite.configuration.annotation.ConfigurationRoot;
+import org.apache.ignite.configuration.annotation.ConfigurationType;
+import org.apache.ignite.configuration.annotation.Value;
+
+/**
+ * Configuration schema for Compute functionality.
+ */
+@ConfigurationRoot(rootName = "deployment", type = ConfigurationType.LOCAL)
+public class DeploymentConfigurationSchema {
+ /**
+ * Relative path to folder in node working directory where will store all
deployment units content.
+ */
+ @Value(hasDefault = true)
+ public final String deploymentLocation = "deployment";
+}
diff --git
a/modules/code-deployment/src/main/java/org/apache/ignite/internal/deployunit/exception/DeploymentUnitAlreadyExistsException.java
b/modules/code-deployment/src/main/java/org/apache/ignite/internal/deployunit/exception/DeploymentUnitAlreadyExistsException.java
new file mode 100644
index 0000000000..726ef2e04f
--- /dev/null
+++
b/modules/code-deployment/src/main/java/org/apache/ignite/internal/deployunit/exception/DeploymentUnitAlreadyExistsException.java
@@ -0,0 +1,48 @@
+/*
+ * 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.deployunit.exception;
+
+import org.apache.ignite.lang.ErrorGroups.CodeDeployment;
+import org.apache.ignite.lang.IgniteException;
+
+/**
+ * Throws when trying to deploy unit which already exist.
+ */
+public class DeploymentUnitAlreadyExistsException extends IgniteException {
+
+ /**
+ * Unit identifier.
+ */
+ private final String id;
+
+ /**
+ * Constructor.
+ *
+ * @param id Unit identifier.
+ * @param message Error message.
+ */
+ public DeploymentUnitAlreadyExistsException(String id, String message) {
+ super(CodeDeployment.UNIT_ALREADY_EXISTS_ERR, message);
+ this.id = id;
+ }
+
+ public String id() {
+ return id;
+ }
+
+}
diff --git
a/modules/code-deployment/src/main/java/org/apache/ignite/internal/deployunit/exception/DeploymentUnitNotFoundException.java
b/modules/code-deployment/src/main/java/org/apache/ignite/internal/deployunit/exception/DeploymentUnitNotFoundException.java
new file mode 100644
index 0000000000..802338b7ca
--- /dev/null
+++
b/modules/code-deployment/src/main/java/org/apache/ignite/internal/deployunit/exception/DeploymentUnitNotFoundException.java
@@ -0,0 +1,35 @@
+/*
+ * 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.deployunit.exception;
+
+import org.apache.ignite.lang.ErrorGroups.CodeDeployment;
+import org.apache.ignite.lang.IgniteException;
+
+/**
+ * Throws when trying to access information about unit which doesn't exist.
+ */
+public class DeploymentUnitNotFoundException extends IgniteException {
+ /**
+ * Constructor.
+ *
+ * @param message error message.
+ */
+ public DeploymentUnitNotFoundException(String message) {
+ super(CodeDeployment.UNIT_NOT_FOUND_ERR, message);
+ }
+}
diff --git
a/modules/code-deployment/src/main/java/org/apache/ignite/internal/deployunit/exception/DeploymentUnitReadException.java
b/modules/code-deployment/src/main/java/org/apache/ignite/internal/deployunit/exception/DeploymentUnitReadException.java
new file mode 100644
index 0000000000..a985769e55
--- /dev/null
+++
b/modules/code-deployment/src/main/java/org/apache/ignite/internal/deployunit/exception/DeploymentUnitReadException.java
@@ -0,0 +1,42 @@
+/*
+ * 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.deployunit.exception;
+
+import org.apache.ignite.lang.ErrorGroups.CodeDeployment;
+import org.apache.ignite.lang.IgniteException;
+
+/**
+ * Throws when deployment unit content failed to read.
+ */
+public class DeploymentUnitReadException extends IgniteException {
+ /**
+ * Constructor.
+ */
+ public DeploymentUnitReadException() {
+ super(CodeDeployment.UNIT_CONTENT_READ_ERR);
+ }
+
+ /**
+ * Constructor.
+ *
+ * @param cause cause error.
+ */
+ public DeploymentUnitReadException(Throwable cause) {
+ super(CodeDeployment.UNIT_CONTENT_READ_ERR, cause);
+ }
+}
diff --git
a/modules/code-deployment/src/main/java/org/apache/ignite/internal/deployunit/message/DeployUnitMessageTypes.java
b/modules/code-deployment/src/main/java/org/apache/ignite/internal/deployunit/message/DeployUnitMessageTypes.java
new file mode 100644
index 0000000000..54337f7fd5
--- /dev/null
+++
b/modules/code-deployment/src/main/java/org/apache/ignite/internal/deployunit/message/DeployUnitMessageTypes.java
@@ -0,0 +1,47 @@
+/*
+ * 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.deployunit.message;
+
+
+import org.apache.ignite.network.annotations.MessageGroup;
+
+/**
+ * Message group for deployment units.
+ */
+@MessageGroup(groupType = 10, groupName = "DeploymentUnit")
+public class DeployUnitMessageTypes {
+ /**
+ * Message type for {@link DeployUnitRequest}.
+ */
+ public static final short DEPLOY_UNIT_REQUEST = 0;
+
+ /**
+ * Message type for {@link DeployUnitResponse}.
+ */
+ public static final short DEPLOY_UNIT_RESPONSE = 1;
+
+ /**
+ * Message type for {@link UndeployUnitRequest}.
+ */
+ public static final short UNDEPLOY_UNIT_REQUEST = 2;
+
+ /**
+ * Message type for {@link UndeployUnitResponse}.
+ */
+ public static final short UNDEPLOY_UNIT_RESPONSE = 3;
+}
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
new file mode 100644
index 0000000000..ddf16bf6d7
--- /dev/null
+++
b/modules/code-deployment/src/main/java/org/apache/ignite/internal/deployunit/message/DeployUnitRequest.java
@@ -0,0 +1,57 @@
+/*
+ * 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.deployunit.message;
+
+import org.apache.ignite.network.NetworkMessage;
+import org.apache.ignite.network.annotations.Transferable;
+
+/**
+ * Deploy unit request.
+ */
+@Transferable(DeployUnitMessageTypes.DEPLOY_UNIT_REQUEST)
+public interface DeployUnitRequest extends NetworkMessage {
+ /**
+ * Returns id of deployment unit.
+ *
+ * @return id of deployment unit
+ */
+ String id();
+
+ /**
+ * Returns version of deployment unit.
+ *
+ * @return version of deployment unit.
+ */
+ String version();
+
+ /**
+ * Returns name of deployment unit.
+ *
+ * @return name of deployment unit.
+ */
+
+ String unitName();
+
+ /**
+ * Returns content of deployment unit.
+ *
+ * @return content of deployment unit.
+ */
+
+ byte[] unitContent();
+}
diff --git
a/modules/code-deployment/src/main/java/org/apache/ignite/internal/deployunit/message/DeployUnitResponse.java
b/modules/code-deployment/src/main/java/org/apache/ignite/internal/deployunit/message/DeployUnitResponse.java
new file mode 100644
index 0000000000..ee298c8a1d
--- /dev/null
+++
b/modules/code-deployment/src/main/java/org/apache/ignite/internal/deployunit/message/DeployUnitResponse.java
@@ -0,0 +1,36 @@
+/*
+ * 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.deployunit.message;
+
+import org.apache.ignite.network.NetworkMessage;
+import org.apache.ignite.network.annotations.Marshallable;
+import org.apache.ignite.network.annotations.Transferable;
+
+/**
+ * Deploy unit response.
+ */
+@Transferable(DeployUnitMessageTypes.DEPLOY_UNIT_RESPONSE)
+public interface DeployUnitResponse extends NetworkMessage {
+ /**
+ * Returns error which happens on deploy process.
+ *
+ * @return error which happens on deploy process.
+ */
+ @Marshallable
+ Throwable error();
+}
diff --git
a/modules/code-deployment/src/main/java/org/apache/ignite/internal/deployunit/message/UndeployUnitRequest.java
b/modules/code-deployment/src/main/java/org/apache/ignite/internal/deployunit/message/UndeployUnitRequest.java
new file mode 100644
index 0000000000..4c4f847670
--- /dev/null
+++
b/modules/code-deployment/src/main/java/org/apache/ignite/internal/deployunit/message/UndeployUnitRequest.java
@@ -0,0 +1,41 @@
+/*
+ * 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.deployunit.message;
+
+import org.apache.ignite.network.NetworkMessage;
+import org.apache.ignite.network.annotations.Transferable;
+
+/**
+ * Undeploy unit request.
+ */
+@Transferable(DeployUnitMessageTypes.UNDEPLOY_UNIT_REQUEST)
+public interface UndeployUnitRequest extends NetworkMessage {
+ /**
+ * Returns id of deployment unit.
+ *
+ * @return id of deployment unit.
+ */
+ String id();
+
+ /**
+ * Returns version of deployment unit.
+ *
+ * @return version of deployment unit.
+ */
+ String version();
+}
diff --git
a/modules/code-deployment/src/main/java/org/apache/ignite/internal/deployunit/message/UndeployUnitResponse.java
b/modules/code-deployment/src/main/java/org/apache/ignite/internal/deployunit/message/UndeployUnitResponse.java
new file mode 100644
index 0000000000..e430724ac4
--- /dev/null
+++
b/modules/code-deployment/src/main/java/org/apache/ignite/internal/deployunit/message/UndeployUnitResponse.java
@@ -0,0 +1,36 @@
+/*
+ * 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.deployunit.message;
+
+import org.apache.ignite.network.NetworkMessage;
+import org.apache.ignite.network.annotations.Marshallable;
+import org.apache.ignite.network.annotations.Transferable;
+
+/**
+ * Undeploy unit response.
+ */
+@Transferable(DeployUnitMessageTypes.UNDEPLOY_UNIT_RESPONSE)
+public interface UndeployUnitResponse extends NetworkMessage {
+ /**
+ * Returns error which happens on undeploy process.
+ *
+ * @return error which happens on undeploy process.
+ */
+ @Marshallable
+ Throwable error();
+}
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
new file mode 100644
index 0000000000..9f4b841110
--- /dev/null
+++
b/modules/code-deployment/src/test/java/org/apache/ignite/deployment/UnitMetaSerializerTest.java
@@ -0,0 +1,82 @@
+/*
+ * 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.deployment;
+
+import static
org.apache.ignite.internal.deployunit.UnitMetaSerializer.deserialize;
+import static
org.apache.ignite.internal.deployunit.UnitMetaSerializer.serialize;
+import static org.hamcrest.Matchers.is;
+
+import java.util.Arrays;
+import java.util.Collections;
+import org.apache.ignite.deployment.version.Version;
+import org.apache.ignite.internal.deployunit.UnitMeta;
+import org.apache.ignite.internal.deployunit.UnitMetaSerializer;
+import org.hamcrest.MatcherAssert;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Test for {@link UnitMetaSerializer}.
+ */
+public class UnitMetaSerializerTest {
+ @Test
+ public void testSerializeDeserializeLatest() {
+ UnitMeta meta = new UnitMeta("id", Version.LATEST, "unitName",
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", 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", 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", Collections.emptyList());
+
+ byte[] serialize = serialize(meta);
+
+ UnitMeta deserialize = deserialize(serialize);
+ MatcherAssert.assertThat(deserialize, is(meta));
+ }
+
+ @Test
+ public void testSerializeDeserializeWithSeparatorCharInIdName() {
+ UnitMeta meta = new UnitMeta("id;", Version.parseVersion("3.0.0"),
"unitName;", Collections.emptyList());
+
+ byte[] serialize = serialize(meta);
+
+ UnitMeta deserialize = deserialize(serialize);
+ MatcherAssert.assertThat(deserialize, is(meta));
+ }
+}
diff --git
a/modules/core/src/main/java/org/apache/ignite/internal/util/IgniteUtils.java
b/modules/core/src/main/java/org/apache/ignite/internal/util/IgniteUtils.java
index 9bdc8f64b2..efa3d4f15e 100644
---
a/modules/core/src/main/java/org/apache/ignite/internal/util/IgniteUtils.java
+++
b/modules/core/src/main/java/org/apache/ignite/internal/util/IgniteUtils.java
@@ -510,26 +510,7 @@ public class IgniteUtils {
*/
public static boolean deleteIfExists(Path path) {
try {
- Files.walkFileTree(path, new SimpleFileVisitor<>() {
- @Override
- public FileVisitResult postVisitDirectory(Path dir,
IOException exc) throws IOException {
- if (exc != null) {
- throw exc;
- }
-
- Files.delete(dir);
-
- return FileVisitResult.CONTINUE;
- }
-
- @Override
- public FileVisitResult visitFile(Path file,
BasicFileAttributes attrs) throws IOException {
- Files.delete(file);
-
- return FileVisitResult.CONTINUE;
- }
- });
-
+ deleteIfExistsThrowable(path);
return true;
} catch (NoSuchFileException e) {
return true;
@@ -538,6 +519,34 @@ public class IgniteUtils {
}
}
+ /**
+ * Deletes a file or a directory with all sub-directories and files.
+ *
+ * @param path File or directory to delete.
+ * @throws IOException if an I/O error is thrown by a visitor method
+ */
+ public static void deleteIfExistsThrowable(Path path) throws IOException {
+ Files.walkFileTree(path, new SimpleFileVisitor<>() {
+ @Override
+ public FileVisitResult postVisitDirectory(Path dir, IOException
exc) throws IOException {
+ if (exc != null) {
+ throw exc;
+ }
+
+ Files.delete(dir);
+
+ return FileVisitResult.CONTINUE;
+ }
+
+ @Override
+ public FileVisitResult visitFile(Path file, BasicFileAttributes
attrs) throws IOException {
+ Files.delete(file);
+
+ return FileVisitResult.CONTINUE;
+ }
+ });
+ }
+
/**
* Checks if assertions enabled.
*
diff --git a/modules/core/src/main/java/org/apache/ignite/lang/ErrorGroups.java
b/modules/core/src/main/java/org/apache/ignite/lang/ErrorGroups.java
index c96b81101b..bed0f9beab 100755
--- a/modules/core/src/main/java/org/apache/ignite/lang/ErrorGroups.java
+++ b/modules/core/src/main/java/org/apache/ignite/lang/ErrorGroups.java
@@ -390,4 +390,29 @@ public class ErrorGroups {
*/
public static final int CONFIG_WRITE_ERR =
NODE_CONFIGURATION_ERR_GROUP.registerErrorCode(3);
}
+
+ /**
+ * Code deployment error group.
+ */
+ public static class CodeDeployment {
+ /**
+ * Code deployment error group.
+ */
+ public static final ErrorGroup CODE_DEPLOYMENT_ERR_GROUP =
ErrorGroup.newGroup("CODEDEPLOY", 13);
+
+ /**
+ * Access to non-existing deployment unit.
+ */
+ public static final int UNIT_NOT_FOUND_ERR =
CODE_DEPLOYMENT_ERR_GROUP.registerErrorCode(1);
+
+ /**
+ * Unit duplicate error.
+ */
+ public static final int UNIT_ALREADY_EXISTS_ERR =
CODE_DEPLOYMENT_ERR_GROUP.registerErrorCode(2);
+
+ /**
+ * Deployment unit content read error.
+ */
+ public static final int UNIT_CONTENT_READ_ERR =
CODE_DEPLOYMENT_ERR_GROUP.registerErrorCode(3);
+ }
}
diff --git a/modules/runner/build.gradle b/modules/runner/build.gradle
index c1484c0bc3..a1f0650691 100644
--- a/modules/runner/build.gradle
+++ b/modules/runner/build.gradle
@@ -55,6 +55,7 @@ dependencies {
implementation project(':ignite-replicator')
implementation project(':ignite-distribution-zones')
implementation project(':ignite-placement-driver')
+ implementation project(':ignite-code-deployment')
implementation libs.jetbrains.annotations
implementation libs.micronaut.inject
implementation libs.micronaut.validation
@@ -114,6 +115,7 @@ dependencies {
integrationTestImplementation project(':ignite-metastorage')
integrationTestImplementation project(':ignite-table')
integrationTestImplementation project(':ignite-transactions')
+ integrationTestImplementation project(':ignite-code-deployment')
integrationTestImplementation project(':ignite-jdbc')
integrationTestImplementation(testFixtures(project(':ignite-core')))
integrationTestImplementation(testFixtures(project(':ignite-configuration')))
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
new file mode 100644
index 0000000000..32659a44dc
--- /dev/null
+++
b/modules/runner/src/integrationTest/java/org/apache/ignite/internal/deployment/ItDeploymentUnitTest.java
@@ -0,0 +1,216 @@
+/*
+ * 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.deployment;
+
+import static java.nio.file.StandardOpenOption.CREATE;
+import static java.nio.file.StandardOpenOption.WRITE;
+import static
org.apache.ignite.internal.testframework.matchers.CompletableFutureMatcher.willBe;
+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;
+import java.nio.channels.SeekableByteChannel;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.util.Collections;
+import java.util.List;
+import java.util.Objects;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+import org.apache.ignite.deployment.DeploymentUnit;
+import org.apache.ignite.deployment.UnitStatus;
+import org.apache.ignite.deployment.UnitStatus.UnitStatusBuilder;
+import org.apache.ignite.deployment.version.Version;
+import org.apache.ignite.internal.AbstractClusterIntegrationTest;
+import org.apache.ignite.internal.app.IgniteImpl;
+import
org.apache.ignite.internal.deployunit.configuration.DeploymentConfiguration;
+import org.apache.ignite.internal.testframework.IgniteTestUtils;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+/**
+ * Integration tests for {@link org.apache.ignite.deployment.IgniteDeployment}.
+ */
+public class ItDeploymentUnitTest extends AbstractClusterIntegrationTest {
+ private static final long REPLICA_TIMEOUT = TimeUnit.SECONDS.toMillis(30);
+ private static final long SIZE_IN_BYTES = 1024L;
+
+ private Path dummyFile;
+
+ @BeforeEach
+ public void generateDummy() throws IOException {
+ dummyFile = workDir.resolve("dummy.txt");
+
+ if (!Files.exists(dummyFile)) {
+ try (SeekableByteChannel channel = Files.newByteChannel(dummyFile,
WRITE, CREATE)) {
+ channel.position(SIZE_IN_BYTES - 4);
+
+ ByteBuffer buf = ByteBuffer.allocate(4).putInt(2);
+ buf.rewind();
+ channel.write(buf);
+ }
+ }
+ }
+
+ @Test
+ public void testDeploy() throws Exception {
+ Unit unit = deployAndVerify("test", Version.parseVersion("1.1.0"), 1);
+
+ IgniteImpl cmg = cluster.node(0);
+ waitUnitReplica(cmg, unit);
+
+ CompletableFuture<List<UnitStatus>> list =
node(2).deployment().unitsAsync();
+ UnitStatusBuilder builder =
UnitStatus.builder(unit.id).append(unit.version,
List.of(unit.deployedNode.name(), cmg.name()));
+ assertThat(list, willBe(List.of(builder.build())));
+ }
+
+ @Test
+ public void testDeployUndeploy() throws Exception {
+ Unit unit = deployAndVerify("test", Version.parseVersion("1.1.0"), 1);
+ unit.undeploy();
+
+ CompletableFuture<List<UnitStatus>> list =
node(2).deployment().unitsAsync();
+ assertThat(list, willBe(Collections.emptyList()));
+ }
+
+ @Test
+ public void testDeployTwoUnits() throws Exception {
+ String id = "test";
+ Unit unit1 = deployAndVerify(id, Version.parseVersion("1.1.0"), 1);
+ Unit unit2 = deployAndVerify(id, Version.parseVersion("1.1.1"), 2);
+
+ IgniteImpl cmg = cluster.node(0);
+ waitUnitReplica(cmg, unit1);
+ waitUnitReplica(cmg, unit2);
+
+ CompletableFuture<UnitStatus> list =
node(2).deployment().statusAsync(id);
+ UnitStatusBuilder status = UnitStatus.builder(id)
+ .append(unit1.version, List.of(unit1.deployedNode.name(),
cmg.name()))
+ .append(unit2.version, List.of(unit2.deployedNode.name(),
cmg.name()));
+ assertThat(list, willBe(status.build()));
+
+ CompletableFuture<List<Version>> versions =
node(2).deployment().versionsAsync(unit1.id);
+ assertThat(versions, willBe(List.of(unit1.version, unit2.version)));
+ }
+
+ @Test
+ public void testDeployTwoUnitsAndUndeployOne() throws Exception {
+ Unit unit1 = deployAndVerify("test", Version.parseVersion("1.1.0"), 1);
+ Unit unit2 = deployAndVerify("test", Version.parseVersion("1.1.1"), 2);
+
+ IgniteImpl cmg = cluster.node(0);
+ waitUnitReplica(cmg, unit1);
+ waitUnitReplica(cmg, unit2);
+
+ CompletableFuture<UnitStatus> list =
node(2).deployment().statusAsync(unit2.id);
+ UnitStatusBuilder builder = UnitStatus.builder(unit1.id)
+ .append(unit1.version, List.of(unit1.deployedNode.name(),
cmg.name()))
+ .append(unit2.version, List.of(unit2.deployedNode.name(),
cmg.name()));
+ assertThat(list, willBe(builder.build()));
+
+ unit2.undeploy();
+ CompletableFuture<List<Version>> newVersions =
node(2).deployment().versionsAsync(unit1.id);
+ assertThat(newVersions, willBe(List.of(unit1.version)));
+ }
+
+ private Unit deployAndVerify(String id, Version version, int nodeIndex) {
+ IgniteImpl entryNode = node(nodeIndex);
+
+ CompletableFuture<Boolean> deploy = entryNode.deployment()
+ .deployAsync(id, version, fromPath(dummyFile));
+
+ assertThat(deploy, willBe(true));
+
+ Unit unit = new Unit(entryNode, id, version);
+ assertTrue(Files.exists(getNodeUnitFile(unit)));
+
+ return unit;
+ }
+
+ private Path getNodeUnitFile(Unit unit) {
+ return getNodeUnitFile(unit.deployedNode, unit.id, unit.version);
+ }
+
+ private Path getNodeUnitFile(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(dummyFile.getFileName());
+ }
+
+ private void waitUnitReplica(IgniteImpl ignite, Unit unit) throws
InterruptedException {
+ Path unitPath = getNodeUnitFile(ignite, unit.id, unit.version);
+ assertTrue(IgniteTestUtils.waitForCondition(() -> {
+ try {
+ return Files.exists(unitPath) && Files.size(unitPath) ==
SIZE_IN_BYTES;
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ }, REPLICA_TIMEOUT));
+ }
+
+ private void waitUnitClean(IgniteImpl ignite, Unit unit) throws
InterruptedException {
+ Path unitPath = getNodeUnitFile(ignite, unit.id, unit.version);
+ assertTrue(IgniteTestUtils.waitForCondition(() ->
!Files.exists(unitPath), REPLICA_TIMEOUT));
+ }
+
+ class Unit {
+ private final IgniteImpl deployedNode;
+
+ private final String id;
+
+ private final Version version;
+
+ Unit(IgniteImpl deployedNode, String id, Version version) {
+ this.deployedNode = deployedNode;
+ this.id = id;
+ this.version = version;
+ }
+
+ void undeploy() throws InterruptedException {
+ deployedNode.deployment().undeployAsync(id, version);
+ waitUnitClean(deployedNode, this);
+ }
+ }
+
+ static DeploymentUnit fromPath(Path path) {
+ Objects.requireNonNull(path);
+ return new DeploymentUnit() {
+
+ @Override
+ public String name() {
+ return path.getFileName().toString();
+ }
+
+ @Override
+ public InputStream content() {
+ try {
+ return new FileInputStream(path.toFile());
+ } catch (IOException e) {
+ throw new RuntimeException(e);
+ }
+ }
+ };
+ }
+}
diff --git
a/modules/runner/src/main/java/org/apache/ignite/internal/app/IgniteImpl.java
b/modules/runner/src/main/java/org/apache/ignite/internal/app/IgniteImpl.java
index c246c3525b..9b015328c9 100644
---
a/modules/runner/src/main/java/org/apache/ignite/internal/app/IgniteImpl.java
+++
b/modules/runner/src/main/java/org/apache/ignite/internal/app/IgniteImpl.java
@@ -42,6 +42,7 @@ import org.apache.ignite.Ignite;
import org.apache.ignite.IgnitionManager;
import org.apache.ignite.client.handler.ClientHandlerModule;
import org.apache.ignite.compute.IgniteCompute;
+import org.apache.ignite.deployment.IgniteDeployment;
import org.apache.ignite.internal.baseline.BaselineManager;
import
org.apache.ignite.internal.cluster.management.ClusterManagementGroupManager;
import
org.apache.ignite.internal.cluster.management.configuration.ClusterManagementConfiguration;
@@ -66,6 +67,8 @@ import
org.apache.ignite.internal.configuration.ServiceLoaderModulesProvider;
import org.apache.ignite.internal.configuration.storage.ConfigurationStorage;
import
org.apache.ignite.internal.configuration.storage.DistributedConfigurationStorage;
import
org.apache.ignite.internal.configuration.storage.LocalFileConfigurationStorage;
+import org.apache.ignite.internal.deployunit.DeploymentManagerImpl;
+import
org.apache.ignite.internal.deployunit.configuration.DeploymentConfiguration;
import org.apache.ignite.internal.distributionzones.DistributionZoneManager;
import
org.apache.ignite.internal.distributionzones.configuration.DistributionZonesConfiguration;
import org.apache.ignite.internal.hlc.HybridClock;
@@ -73,6 +76,7 @@ import org.apache.ignite.internal.hlc.HybridClockImpl;
import org.apache.ignite.internal.index.IndexManager;
import org.apache.ignite.internal.logger.IgniteLogger;
import org.apache.ignite.internal.logger.Loggers;
+import org.apache.ignite.internal.manager.IgniteComponent;
import org.apache.ignite.internal.metastorage.MetaStorageManager;
import org.apache.ignite.internal.metastorage.impl.MetaStorageManagerImpl;
import
org.apache.ignite.internal.metastorage.server.persistence.RocksDbKeyValueStorage;
@@ -249,6 +253,8 @@ public class IgniteImpl implements Ignite {
/** Metric manager. */
private final MetricManager metricManager;
+ private final IgniteDeployment deploymentManager;
+
private final DistributionZoneManager distributionZoneManager;
/** Creator for volatile {@link
org.apache.ignite.internal.raft.storage.LogStorageFactory} instances. */
@@ -497,6 +503,12 @@ public class IgniteImpl implements Ignite {
sql,
() -> cmgMgr.clusterState().thenApply(s ->
s.clusterTag().clusterId())
);
+
+ deploymentManager = new DeploymentManagerImpl(clusterSvc,
+ metaStorageMgr,
+ workDir,
+
nodeConfigRegistry.getConfiguration(DeploymentConfiguration.KEY),
+ cmgMgr);
}
private RestComponent createRestComponent(String name) {
@@ -627,7 +639,8 @@ public class IgniteImpl implements Ignite {
distributedTblMgr,
indexManager,
qryEngine,
- clientHandlerModule
+ clientHandlerModule,
+ (IgniteComponent) deploymentManager
);
} catch (NodeStoppingException e) {
throw new CompletionException(e);
@@ -750,6 +763,11 @@ public class IgniteImpl implements Ignite {
return compute;
}
+ @Override
+ public IgniteDeployment deployment() {
+ return deploymentManager;
+ }
+
/** {@inheritDoc} */
@Override
public Collection<ClusterNode> clusterNodes() {
diff --git a/settings.gradle b/settings.gradle
index 5ee56ae73d..3632092a4a 100644
--- a/settings.gradle
+++ b/settings.gradle
@@ -65,6 +65,7 @@ include(':packaging')
include(':ignite-replicator')
include(':ignite-distribution-zones')
include(':ignite-placement-driver')
+include(':ignite-code-deployment')
project(":ignite-examples").projectDir = file('examples')
project(":ignite-page-memory").projectDir = file('modules/page-memory')
@@ -115,6 +116,7 @@ project(":packaging-db").projectDir = file('packaging/db')
project(":packaging").projectDir = file('packaging')
project(":ignite-distribution-zones").projectDir =
file('modules/distribution-zones')
project(":ignite-placement-driver").projectDir =
file('modules/placement-driver')
+project(":ignite-code-deployment").projectDir = file('modules/code-deployment')
ext.isCiServer = System.getenv().containsKey("IGNITE_CI")
