ibessonov commented on code in PR #1698: URL: https://github.com/apache/ignite-3/pull/1698#discussion_r1114024749
########## 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 -> { Review Comment: Ok, so you create a list of nodes, and the only usage for it is "iteration". Streams were designed with other intention in mind. For example, you can use "forEach", in that case no new list would be created, making you code faster and more "idiomatic", I guess. Maybe there are hidden reasons for having an explicit list, that I can't understand right now. In that case, please express your reasoning in the comment. -- 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]
