ibessonov commented on code in PR #1698: URL: https://github.com/apache/ignite-3/pull/1698#discussion_r1114040292
########## modules/code-deployment/src/main/java/org/apache/ignite/internal/deployunit/DeploymentManagerImpl.java: ########## @@ -0,0 +1,424 @@ +/* + * 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.FileVisitResult; +import java.nio.file.Files; +import java.nio.file.Path; +import java.nio.file.SimpleFileVisitor; +import java.nio.file.StandardCopyOption; +import java.nio.file.StandardOpenOption; +import java.nio.file.attribute.BasicFileAttributes; +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Set; +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.DeployUnitWriteMetaException; +import org.apache.ignite.internal.deployunit.exception.DeploymentUnitIdentifierException; +import org.apache.ignite.internal.deployunit.exception.DeploymentUnitNotExistException; +import org.apache.ignite.internal.deployunit.exception.DeploymentUnitReadException; +import org.apache.ignite.internal.deployunit.exception.UndeployNotExistedDeploymentUnitException; +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.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 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) { + if (id == null || id.isBlank() || version == null) { + return CompletableFuture.failedFuture(new DeploymentUnitIdentifierException()); + } + + 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 DeployUnitWriteMetaException()); + }) + .thenApply(completed -> { + if (completed) { + startDeployAsyncToCmg(request); + } + return completed; + }); + } + + private void startDeployAsyncToCmg(DeployUnitRequest request) { + cmgManager.cmgNodes() + .thenApply(nodes -> nodes.stream().map(node -> clusterService.topologyService().getByConsistentId(node)).collect( + Collectors.toList())) + .thenAccept(clusterNodes -> { + for (ClusterNode clusterNode : clusterNodes) { + inFlightFutures.registerFuture( + 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) { + if (id == null || id.isBlank() || version == null) { + return CompletableFuture.failedFuture(new DeploymentUnitIdentifierException()); + } + + 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 UndeployNotExistedDeploymentUnitException( + "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); + } + return null; + }); + } + + @Override + public CompletableFuture<List<UnitStatus>> listAsync() { + 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.getId(), UnitStatus::builder) + .append(meta.getVersion(), meta.getConsistentIdLocation()); + } + + @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) { + if (id == null || id.isBlank()) { + return CompletableFuture.failedFuture(new DeploymentUnitIdentifierException()); + } + CompletableFuture<List<Version>> result = new CompletableFuture<>(); + metaStorage.prefix(new ByteArray(UNITS_PREFIX + id)) + .subscribe(new Subscriber<>() { + private final List<Version> set = 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()); + set.add(deserialize.getVersion()); + } + + @Override + public void onError(Throwable throwable) { + result.completeExceptionally(throwable); + } + + @Override + public void onComplete() { + result.complete(set); + } + }); + return result; + } + + @Override + public CompletableFuture<UnitStatus> statusAsync(String id) { + if (id == null || id.isBlank()) { + return CompletableFuture.failedFuture(new DeploymentUnitIdentifierException()); + } + 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.getVersion(), deserialize.getConsistentIdLocation()); + } + + @Override + public void onError(Throwable throwable) { + result.completeExceptionally(throwable); + } + + @Override + public void onComplete() { + if (builder != null) { + result.complete(builder.build()); + } else { + result.completeExceptionally( + new DeploymentUnitNotExistException("Unit with " + id + "not exist.")); Review Comment: First of all, there should be a space before "not exist." Second, "not exist" is grammatically incorrect, it should be "doesn't exist". -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
