http://git-wip-us.apache.org/repos/asf/ignite/blob/62c560a5/modules/core/src/main/java/org/apache/ignite/internal/processors/service/IgniteServiceProcessor.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/service/IgniteServiceProcessor.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/service/IgniteServiceProcessor.java new file mode 100644 index 0000000..1bc1fd7 --- /dev/null +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/service/IgniteServiceProcessor.java @@ -0,0 +1,1784 @@ +/* + * 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.processors.service; + +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collection; +import java.util.Collections; +import java.util.HashMap; +import java.util.HashSet; +import java.util.Iterator; +import java.util.List; +import java.util.Map; +import java.util.Random; +import java.util.Set; +import java.util.TreeMap; +import java.util.TreeSet; +import java.util.UUID; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.ConcurrentMap; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.ThreadFactory; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.locks.ReentrantReadWriteLock; +import java.util.stream.Collectors; +import java.util.stream.Stream; +import org.apache.ignite.IgniteCheckedException; +import org.apache.ignite.IgniteException; +import org.apache.ignite.cluster.ClusterGroup; +import org.apache.ignite.cluster.ClusterNode; +import org.apache.ignite.configuration.DeploymentMode; +import org.apache.ignite.configuration.IgniteConfiguration; +import org.apache.ignite.events.DiscoveryEvent; +import org.apache.ignite.internal.GridKernalContext; +import org.apache.ignite.internal.IgniteClientDisconnectedCheckedException; +import org.apache.ignite.internal.IgniteInternalFuture; +import org.apache.ignite.internal.IgniteInterruptedCheckedException; +import org.apache.ignite.internal.SkipDaemon; +import org.apache.ignite.internal.managers.discovery.CustomEventListener; +import org.apache.ignite.internal.managers.discovery.DiscoCache; +import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion; +import org.apache.ignite.internal.processors.cache.DynamicCacheChangeBatch; +import org.apache.ignite.internal.processors.cache.DynamicCacheChangeRequest; +import org.apache.ignite.internal.processors.cluster.ChangeGlobalStateMessage; +import org.apache.ignite.internal.processors.cluster.DiscoveryDataClusterState; +import org.apache.ignite.internal.processors.cluster.IgniteChangeGlobalStateSupport; +import org.apache.ignite.internal.util.future.GridCompoundFuture; +import org.apache.ignite.internal.util.future.GridFinishedFuture; +import org.apache.ignite.internal.util.future.GridFutureAdapter; +import org.apache.ignite.internal.util.typedef.F; +import org.apache.ignite.internal.util.typedef.internal.A; +import org.apache.ignite.internal.util.typedef.internal.U; +import org.apache.ignite.lang.IgniteFuture; +import org.apache.ignite.lang.IgnitePredicate; +import org.apache.ignite.lang.IgniteUuid; +import org.apache.ignite.marshaller.Marshaller; +import org.apache.ignite.marshaller.jdk.JdkMarshaller; +import org.apache.ignite.plugin.security.SecurityPermission; +import org.apache.ignite.services.Service; +import org.apache.ignite.services.ServiceConfiguration; +import org.apache.ignite.services.ServiceDeploymentException; +import org.apache.ignite.services.ServiceDescriptor; +import org.apache.ignite.spi.communication.CommunicationSpi; +import org.apache.ignite.spi.discovery.DiscoveryDataBag; +import org.apache.ignite.spi.discovery.DiscoverySpi; +import org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi; +import org.apache.ignite.thread.IgniteThreadFactory; +import org.apache.ignite.thread.OomExceptionHandler; +import org.jetbrains.annotations.NotNull; +import org.jetbrains.annotations.Nullable; + +import static org.apache.ignite.configuration.DeploymentMode.ISOLATED; +import static org.apache.ignite.configuration.DeploymentMode.PRIVATE; +import static org.apache.ignite.events.EventType.EVT_NODE_JOINED; +import static org.apache.ignite.internal.GridComponent.DiscoveryDataExchangeType.SERVICE_PROC; + +/** + * Ignite service processor. + * <p/> + * Event-driven implementation of the service processor. Service deployment is managed via {@link DiscoverySpi} and + * {@link CommunicationSpi} messages. + * + * @see ServiceDeploymentManager + * @see ServiceDeploymentTask + * @see ServiceDeploymentActions + * @see ServiceChangeBatchRequest + */ +@SkipDaemon +@SuppressWarnings("SynchronizationOnLocalVariableOrMethodParameter") +public class IgniteServiceProcessor extends ServiceProcessorAdapter implements IgniteChangeGlobalStateSupport { + /** Local service instances. */ + private final ConcurrentMap<IgniteUuid, Collection<ServiceContextImpl>> locServices = new ConcurrentHashMap<>(); + + /** + * Collection of services information that were registered in the cluster. <b>It is updated from discovery + * thread</b>. It will be included in the initial data bag to be sent on newly joining node, it means a new node + * will have all services' information to be able to work with services in the whole cluster. + * <p/> + * This collection is used, to make fast verification for requests of change service's state and prepare services + * deployments actions (possible long-running) which will be processed from a queue by deployment worker. + * <p/> + * Collection reflects a services' state which will be reached as soon as a relevant deployment task will be + * processed. + * + * @see ServiceDeploymentActions + * @see ServiceDeploymentManager + */ + private final ConcurrentMap<IgniteUuid, ServiceInfo> registeredServices = new ConcurrentHashMap<>(); + + /** + * Collection of services information that were processed by deployment worker. <b>It is updated from deployment + * worker</b>. + * <p/> + * Collection reflects a state of deployed services for a moment of the latest deployment task processed by + * deployment worker. + * <p/> + * It is catching up the state of {@link #registeredServices}. + * + * @see ServiceDeploymentManager#readyTopologyVersion() + * @see ServiceDeploymentTask + */ + private final ConcurrentMap<IgniteUuid, ServiceInfo> deployedServices = new ConcurrentHashMap<>(); + + /** Deployment futures. */ + private final ConcurrentMap<IgniteUuid, GridServiceDeploymentFuture<IgniteUuid>> depFuts = new ConcurrentHashMap<>(); + + /** Undeployment futures. */ + private final ConcurrentMap<IgniteUuid, GridFutureAdapter<?>> undepFuts = new ConcurrentHashMap<>(); + + /** Thread factory. */ + private final ThreadFactory threadFactory = new IgniteThreadFactory(ctx.igniteInstanceName(), "service", + new OomExceptionHandler(ctx)); + + /** Marshaller for serialization/deserialization of service's instance. */ + private final Marshaller marsh = new JdkMarshaller(); + + /** Services deployment manager. */ + private volatile ServiceDeploymentManager depMgr = new ServiceDeploymentManager(ctx); + + /** Services topologies update mutex. */ + private final Object servicesTopsUpdateMux = new Object(); + + /** + * Operations lock. The main purpose is to avoid a hang of users operation futures. + * <p/> + * Read lock is being acquired on users operations (deploy, cancel). + * <p/> + * Write lock is being acquired on change service processor's state: {@link #onKernalStop}, {@link #onDisconnected), + * {@link #onDeActivate(GridKernalContext)}} to guarantee that deployed services will be cancelled only once, also + * it protects from registering new operations futures which may be missed during completion collections of users + * futures. + * <pre> + * {@link #enterBusy()} and {@link #leaveBusy()} are being used to protect modification of shared collections during + * changing service processor state. If a call can't enter in the busy state a default value will be returned (a + * value which will be reached by the time when write lock will be released). + * These methods can't be used for users operations (deploy, undeploy) because if the processor will become + * disconnected or stopped we should return different types of exceptions (it's not about just a errors message, + * the disconnected exception also contains reconnect future). + * </pre> + */ + private final ReentrantReadWriteLock opsLock = new ReentrantReadWriteLock(); + + /** Disconnected flag. */ + private volatile boolean disconnected; + + /** + * @param ctx Kernal context. + */ + public IgniteServiceProcessor(GridKernalContext ctx) { + super(ctx); + } + + /** {@inheritDoc} */ + @Override public void start() throws IgniteCheckedException { + IgniteConfiguration cfg = ctx.config(); + + DeploymentMode depMode = cfg.getDeploymentMode(); + + if (cfg.isPeerClassLoadingEnabled() && (depMode == PRIVATE || depMode == ISOLATED) && + !F.isEmpty(cfg.getServiceConfiguration())) + throw new IgniteCheckedException("Cannot deploy services in PRIVATE or ISOLATED deployment mode: " + depMode); + + ctx.discovery().setCustomEventListener(ServiceChangeBatchRequest.class, + new CustomEventListener<ServiceChangeBatchRequest>() { + @Override public void onCustomEvent(AffinityTopologyVersion topVer, ClusterNode snd, + ServiceChangeBatchRequest msg) { + processServicesChangeRequest(snd, msg); + } + }); + + ctx.discovery().setCustomEventListener(ChangeGlobalStateMessage.class, + new CustomEventListener<ChangeGlobalStateMessage>() { + @Override public void onCustomEvent(AffinityTopologyVersion topVer, ClusterNode snd, + ChangeGlobalStateMessage msg) { + processChangeGlobalStateRequest(msg); + } + }); + + ctx.discovery().setCustomEventListener(DynamicCacheChangeBatch.class, + new CustomEventListener<DynamicCacheChangeBatch>() { + @Override public void onCustomEvent(AffinityTopologyVersion topVer, ClusterNode snd, + DynamicCacheChangeBatch msg) { + processDynamicCacheChangeRequest(msg); + } + }); + + ctx.discovery().setCustomEventListener(ServiceClusterDeploymentResultBatch.class, + new CustomEventListener<ServiceClusterDeploymentResultBatch>() { + @Override public void onCustomEvent(AffinityTopologyVersion topVer, ClusterNode snd, + ServiceClusterDeploymentResultBatch msg) { + processServicesFullDeployments(msg); + } + }); + } + + /** {@inheritDoc} */ + @Override public void onKernalStart(boolean active) throws IgniteCheckedException { + depMgr.startProcessing(); + + if (log.isDebugEnabled()) + log.debug("Started service processor."); + } + + /** {@inheritDoc} */ + @Override public void onKernalStop(boolean cancel) { + opsLock.writeLock().lock(); + + try { + if (disconnected) + return; + + stopProcessor(new IgniteCheckedException("Operation has been cancelled (node is stopping).")); + } + finally { + opsLock.writeLock().unlock(); + } + } + + /** + * @param stopError Error to shutdown resources. + */ + private void stopProcessor(IgniteCheckedException stopError) { + assert opsLock.isWriteLockedByCurrentThread(); + + depMgr.stopProcessing(stopError); + + cancelDeployedServices(); + + registeredServices.clear(); + + // If user requests sent to network but not received back to handle in deployment manager. + Stream.concat(depFuts.values().stream(), undepFuts.values().stream()).forEach(fut -> { + try { + fut.onDone(stopError); + } + catch (Exception ignore) { + // No-op. + } + }); + + depFuts.clear(); + undepFuts.clear(); + + if (log.isDebugEnabled()) + log.debug("Stopped service processor."); + } + + /** + * Cancels deployed services. + */ + private void cancelDeployedServices() { + assert opsLock.isWriteLockedByCurrentThread(); + + deployedServices.clear(); + + locServices.values().stream().flatMap(Collection::stream).forEach(srvcCtx -> { + cancel(srvcCtx); + + if (ctx.isStopping()) { + try { + if (log.isInfoEnabled()) { + log.info("Shutting down distributed service [name=" + srvcCtx.name() + ", execId8=" + + U.id8(srvcCtx.executionId()) + ']'); + } + + srvcCtx.executor().awaitTermination(Long.MAX_VALUE, TimeUnit.MILLISECONDS); + } + catch (InterruptedException ignore) { + Thread.currentThread().interrupt(); + + U.error(log, "Got interrupted while waiting for service to shutdown (will continue " + + "stopping node): " + srvcCtx.name()); + } + } + }); + + locServices.clear(); + } + + /** {@inheritDoc} */ + @Override public void collectGridNodeData(DiscoveryDataBag dataBag) { + if (dataBag.commonDataCollectedFor(SERVICE_PROC.ordinal())) + return; + + ServiceProcessorCommonDiscoveryData clusterData = new ServiceProcessorCommonDiscoveryData( + new ArrayList<>(registeredServices.values()) + ); + + dataBag.addGridCommonData(SERVICE_PROC.ordinal(), clusterData); + } + + /** {@inheritDoc} */ + @Override public void onGridDataReceived(DiscoveryDataBag.GridDiscoveryData data) { + if (data.commonData() == null) + return; + + ServiceProcessorCommonDiscoveryData clusterData = (ServiceProcessorCommonDiscoveryData)data.commonData(); + + for (ServiceInfo desc : clusterData.registeredServices()) + registeredServices.put(desc.serviceId(), desc); + } + + /** {@inheritDoc} */ + @Override public void collectJoiningNodeData(DiscoveryDataBag dataBag) { + ArrayList<ServiceInfo> staticServicesInfo = staticallyConfiguredServices(true); + + dataBag.addJoiningNodeData(SERVICE_PROC.ordinal(), new ServiceProcessorJoinNodeDiscoveryData(staticServicesInfo)); + } + + /** {@inheritDoc} */ + @Override public void onJoiningNodeDataReceived(DiscoveryDataBag.JoiningNodeDiscoveryData data) { + if (data.joiningNodeData() == null) + return; + + ServiceProcessorJoinNodeDiscoveryData joinData = (ServiceProcessorJoinNodeDiscoveryData)data.joiningNodeData(); + + for (ServiceInfo desc : joinData.services()) { + assert desc.topologySnapshot().isEmpty(); + + ServiceInfo oldDesc = registeredServices.get(desc.serviceId()); + + if (oldDesc != null) { // In case of a collision of IgniteUuid.randomUuid() (almost impossible case) + U.warn(log, "Failed to register service configuration received from joining node : " + + "[nodeId=" + data.joiningNodeId() + ", cfgName=" + desc.name() + "]. " + + "Service with the same service id already exists, cfg=" + oldDesc.configuration()); + + continue; + } + + oldDesc = lookupInRegisteredServices(desc.name()); + + if (oldDesc == null) { + registeredServices.put(desc.serviceId(), desc); + + continue; + } + + if (oldDesc.configuration().equalsIgnoreNodeFilter(desc.configuration())) { + if (log.isDebugEnabled()) { + log.debug("Ignore service configuration received from joining node : " + + "[nodeId=" + data.joiningNodeId() + ", cfgName=" + desc.name() + "]. " + + "The same service configuration already registered."); + } + } + else { + U.warn(log, "Failed to register service configuration received from joining node : " + + "[nodeId=" + data.joiningNodeId() + ", cfgName=" + desc.name() + "]. " + + "Service already exists with different configuration, cfg=" + desc.configuration()); + } + } + } + + /** {@inheritDoc} */ + @Nullable @Override public DiscoveryDataExchangeType discoveryDataType() { + return SERVICE_PROC; + } + + /** {@inheritDoc} */ + @Override public void onActivate(GridKernalContext kctx) { + // No-op. + } + + /** + * Invokes from services deployment worker. + * <p/> + * {@inheritDoc} + */ + @Override public void onDeActivate(GridKernalContext kctx) { + opsLock.writeLock().lock(); + + try { + if (log.isDebugEnabled()) { + log.debug("DeActivate service processor [nodeId=" + ctx.localNodeId() + + " topVer=" + ctx.discovery().topologyVersionEx() + " ]"); + } + + cancelDeployedServices(); + } + finally { + opsLock.writeLock().unlock(); + } + } + + /** {@inheritDoc} */ + @Override public void onDisconnected(IgniteFuture<?> reconnectFut) { + assert !disconnected; + + opsLock.writeLock().lock(); + + try { + disconnected = true; + + stopProcessor(new IgniteClientDisconnectedCheckedException( + ctx.cluster().clientReconnectFuture(), "Client node disconnected, the operation's result is unknown.")); + } + finally { + opsLock.writeLock().unlock(); + } + } + + /** {@inheritDoc} */ + @Override public IgniteInternalFuture<?> onReconnected(boolean active) throws IgniteCheckedException { + assert disconnected; + + opsLock.writeLock().lock(); + + try { + disconnected = false; + + depMgr = new ServiceDeploymentManager(ctx); + + onKernalStart(active); + + return null; + } + finally { + opsLock.writeLock().unlock(); + } + } + + /** + * Validates service configuration. + * + * @param c Service configuration. + * @throws IgniteException If validation failed. + */ + private void validate(ServiceConfiguration c) throws IgniteException { + IgniteConfiguration cfg = ctx.config(); + + DeploymentMode depMode = cfg.getDeploymentMode(); + + if (cfg.isPeerClassLoadingEnabled() && (depMode == PRIVATE || depMode == ISOLATED)) + throw new IgniteException("Cannot deploy services in PRIVATE or ISOLATED deployment mode: " + depMode); + + ensure(c.getName() != null, "getName() != null", null); + ensure(c.getTotalCount() >= 0, "getTotalCount() >= 0", c.getTotalCount()); + ensure(c.getMaxPerNodeCount() >= 0, "getMaxPerNodeCount() >= 0", c.getMaxPerNodeCount()); + ensure(c.getService() != null, "getService() != null", c.getService()); + ensure(c.getTotalCount() > 0 || c.getMaxPerNodeCount() > 0, + "c.getTotalCount() > 0 || c.getMaxPerNodeCount() > 0", null); + } + + /** + * @param cond Condition. + * @param desc Description. + * @param v Value. + */ + private void ensure(boolean cond, String desc, @Nullable Object v) { + if (!cond) + if (v != null) + throw new IgniteException("Service configuration check failed (" + desc + "): " + v); + else + throw new IgniteException("Service configuration check failed (" + desc + ")"); + } + + /** {@inheritDoc} */ + @Override public IgniteInternalFuture<?> deployNodeSingleton(ClusterGroup prj, String name, Service srvc) { + return deployMultiple(prj, name, srvc, 0, 1); + } + + /** {@inheritDoc} */ + @Override public IgniteInternalFuture<?> deployClusterSingleton(ClusterGroup prj, String name, Service srvc) { + return deployMultiple(prj, name, srvc, 1, 1); + } + + /** {@inheritDoc} */ + @Override public IgniteInternalFuture<?> deployMultiple(ClusterGroup prj, String name, Service srvc, int totalCnt, + int maxPerNodeCnt) { + ServiceConfiguration cfg = new ServiceConfiguration(); + + cfg.setName(name); + cfg.setService(srvc); + cfg.setTotalCount(totalCnt); + cfg.setMaxPerNodeCount(maxPerNodeCnt); + + return deployAll(prj, Collections.singleton(cfg)); + } + + /** {@inheritDoc} */ + @Override public IgniteInternalFuture<?> deployKeyAffinitySingleton(String name, Service srvc, String cacheName, + Object affKey) { + A.notNull(affKey, "affKey"); + + ServiceConfiguration cfg = new ServiceConfiguration(); + + cfg.setName(name); + cfg.setService(srvc); + cfg.setCacheName(cacheName); + cfg.setAffinityKey(affKey); + cfg.setTotalCount(1); + cfg.setMaxPerNodeCount(1); + + // Ignore projection here. + return deployAll(Collections.singleton(cfg), null); + } + + /** + * @param cfgs Service configurations. + * @param dfltNodeFilter Default NodeFilter. + * @return Configurations to deploy. + */ + @SuppressWarnings("deprecation") + private PreparedConfigurations<IgniteUuid> prepareServiceConfigurations(Collection<ServiceConfiguration> cfgs, + IgnitePredicate<ClusterNode> dfltNodeFilter) { + List<ServiceConfiguration> cfgsCp = new ArrayList<>(cfgs.size()); + + List<GridServiceDeploymentFuture<IgniteUuid>> failedFuts = null; + + for (ServiceConfiguration cfg : cfgs) { + Exception err = null; + + // Deploy to projection node by default + // or only on server nodes if no projection. + if (cfg.getNodeFilter() == null && dfltNodeFilter != null) + cfg.setNodeFilter(dfltNodeFilter); + + try { + validate(cfg); + } + catch (Exception e) { + U.error(log, "Failed to validate service configuration [name=" + cfg.getName() + + ", srvc=" + cfg.getService() + ']', e); + + err = e; + } + + if (err == null) + err = checkPermissions(cfg.getName(), SecurityPermission.SERVICE_DEPLOY); + + if (err == null) { + try { + byte[] srvcBytes = U.marshal(marsh, cfg.getService()); + + cfgsCp.add(new LazyServiceConfiguration(cfg, srvcBytes)); + } + catch (Exception e) { + U.error(log, "Failed to marshal service with configured marshaller " + + "[name=" + cfg.getName() + ", srvc=" + cfg.getService() + ", marsh=" + marsh + "]", e); + + err = e; + } + } + + if (err != null) { + if (failedFuts == null) + failedFuts = new ArrayList<>(); + + GridServiceDeploymentFuture<IgniteUuid> fut = new GridServiceDeploymentFuture<>(cfg, null); + + fut.onDone(err); + + failedFuts.add(fut); + } + } + + return new PreparedConfigurations<>(cfgsCp, failedFuts); + } + + /** + * Checks security permissions for service with given name. + * + * @param name Service name. + * @param perm Security permissions. + * @return {@code null} if success, otherwise instance of {@link SecurityException}. + */ + private SecurityException checkPermissions(String name, SecurityPermission perm) { + try { + ctx.security().authorize(name, perm, null); + + return null; + } + catch (SecurityException e) { + U.error(log, "Failed to authorize service access [name=" + name + ", perm=" + perm + ']', e); + + return e; + } + } + + /** {@inheritDoc} */ + @Override public IgniteInternalFuture<?> deployAll(ClusterGroup prj, Collection<ServiceConfiguration> cfgs) { + if (prj == null) + // Deploy to servers by default if no projection specified. + return deployAll(cfgs, ctx.cluster().get().forServers().predicate()); + else if (prj.predicate() == F.<ClusterNode>alwaysTrue()) + return deployAll(cfgs, null); + else + // Deploy to predicate nodes by default. + return deployAll(cfgs, prj.predicate()); + } + + /** + * @param cfgs Service configurations. + * @param dfltNodeFilter Default NodeFilter. + * @return Future for deployment. + */ + private IgniteInternalFuture<?> deployAll(@NotNull Collection<ServiceConfiguration> cfgs, + @Nullable IgnitePredicate<ClusterNode> dfltNodeFilter) { + opsLock.readLock().lock(); + + try { + if (disconnected) { + return new GridFinishedFuture<>(new IgniteClientDisconnectedCheckedException( + ctx.cluster().clientReconnectFuture(), "Failed to deploy services, " + + "client node disconnected: " + cfgs)); + } + + if (ctx.isStopping()) { + return new GridFinishedFuture<>(new IgniteCheckedException("Failed to deploy services, " + + "node is stopping: " + cfgs)); + } + + if (cfgs.isEmpty()) + return new GridFinishedFuture<>(); + + PreparedConfigurations<IgniteUuid> srvcCfg = prepareServiceConfigurations(cfgs, dfltNodeFilter); + + List<ServiceConfiguration> cfgsCp = srvcCfg.cfgs; + + List<GridServiceDeploymentFuture<IgniteUuid>> failedFuts = srvcCfg.failedFuts; + + GridServiceDeploymentCompoundFuture<IgniteUuid> res = new GridServiceDeploymentCompoundFuture<>(); + + if (!cfgsCp.isEmpty()) { + try { + Collection<ServiceChangeAbstractRequest> reqs = new ArrayList<>(); + + for (ServiceConfiguration cfg : cfgsCp) { + IgniteUuid srvcId = IgniteUuid.randomUuid(); + + GridServiceDeploymentFuture<IgniteUuid> fut = new GridServiceDeploymentFuture<>(cfg, srvcId); + + res.add(fut, true); + + reqs.add(new ServiceDeploymentRequest(srvcId, cfg)); + + depFuts.put(srvcId, fut); + } + + ServiceChangeBatchRequest msg = new ServiceChangeBatchRequest(reqs); + + ctx.discovery().sendCustomEvent(msg); + + if (log.isDebugEnabled()) + log.debug("Services have been sent to deploy, req=" + msg); + } + catch (IgniteException | IgniteCheckedException e) { + for (IgniteUuid id : res.servicesToRollback()) + depFuts.remove(id).onDone(e); + + res.onDone(new IgniteCheckedException( + new ServiceDeploymentException("Failed to deploy provided services.", e, cfgs))); + + return res; + } + } + + if (failedFuts != null) { + for (GridServiceDeploymentFuture<IgniteUuid> fut : failedFuts) + res.add(fut, false); + } + + res.markInitialized(); + + return res; + } + finally { + opsLock.readLock().unlock(); + } + } + + /** {@inheritDoc} */ + @Override public IgniteInternalFuture<?> cancel(String name) { + return cancelAll(Collections.singleton(name)); + } + + /** {@inheritDoc} */ + @Override public IgniteInternalFuture<?> cancelAll() { + return cancelAll(deployedServices.values().stream().map(ServiceInfo::name).collect(Collectors.toSet())); + } + + /** {@inheritDoc} */ + @SuppressWarnings("unchecked") + @Override public IgniteInternalFuture<?> cancelAll(@NotNull Collection<String> servicesNames) { + opsLock.readLock().lock(); + + try { + if (disconnected) { + return new GridFinishedFuture<>(new IgniteClientDisconnectedCheckedException( + ctx.cluster().clientReconnectFuture(), "Failed to undeploy services, " + + "client node disconnected: " + servicesNames)); + } + + if (ctx.isStopping()) { + return new GridFinishedFuture<>(new IgniteCheckedException("Failed to undeploy services, " + + "node is stopping: " + servicesNames)); + } + + if (servicesNames.isEmpty()) + return new GridFinishedFuture<>(); + + GridCompoundFuture res = new GridCompoundFuture<>(); + + Set<IgniteUuid> toRollback = new HashSet<>(); + + List<ServiceChangeAbstractRequest> reqs = new ArrayList<>(); + + try { + for (String name : servicesNames) { + IgniteUuid srvcId = lookupDeployedServiceId(name); + + if (srvcId == null) + continue; + + Exception err = checkPermissions(name, SecurityPermission.SERVICE_CANCEL); + + if (err != null) { + res.add(new GridFinishedFuture<>(err)); + + continue; + } + + GridFutureAdapter<?> fut = new GridFutureAdapter<>(); + + GridFutureAdapter<?> old = undepFuts.putIfAbsent(srvcId, fut); + + if (old != null) { + res.add(old); + + continue; + } + + res.add(fut); + + toRollback.add(srvcId); + + reqs.add(new ServiceUndeploymentRequest(srvcId)); + } + + if (!reqs.isEmpty()) { + ServiceChangeBatchRequest msg = new ServiceChangeBatchRequest(reqs); + + ctx.discovery().sendCustomEvent(msg); + + if (log.isDebugEnabled()) + log.debug("Services have been sent to cancel, msg=" + msg); + } + } + catch (IgniteException | IgniteCheckedException e) { + for (IgniteUuid id : toRollback) + undepFuts.remove(id).onDone(e); + + U.error(log, "Failed to undeploy services: " + servicesNames, e); + + res.onDone(e); + + return res; + } + + res.markInitialized(); + + return res; + } + finally { + opsLock.readLock().unlock(); + } + } + + /** {@inheritDoc} */ + @Override public Map<UUID, Integer> serviceTopology(String name, long timeout) throws IgniteCheckedException { + assert timeout >= 0; + + long startTime = U.currentTimeMillis(); + + Map<UUID, Integer> top; + + while (true) { + top = serviceTopology(name); + + if (timeout == 0 || (top != null && !top.isEmpty())) + return top; + + synchronized (servicesTopsUpdateMux) { + long wait = timeout - (U.currentTimeMillis() - startTime); + + if (wait <= 0) + return top; + + try { + servicesTopsUpdateMux.wait(wait); + } + catch (InterruptedException e) { + throw new IgniteInterruptedCheckedException(e); + } + } + } + } + + /** + * @param name Service name. + * @return Service topology. + */ + private Map<UUID, Integer> serviceTopology(String name) { + for (ServiceInfo desc : registeredServices.values()) { + if (desc.name().equals(name)) + return desc.topologySnapshot(); + } + + return null; + } + + /** {@inheritDoc} */ + @Override public Collection<ServiceDescriptor> serviceDescriptors() { + return new ArrayList<>(registeredServices.values()); + } + + /** {@inheritDoc} */ + @Override public <T> T service(String name) { + if (!enterBusy()) + return null; + + try { + ctx.security().authorize(name, SecurityPermission.SERVICE_INVOKE, null); + + Collection<ServiceContextImpl> ctxs = serviceContexts(name); + + if (ctxs == null) + return null; + + synchronized (ctxs) { + if (F.isEmpty(ctxs)) + return null; + + for (ServiceContextImpl ctx : ctxs) { + Service srvc = ctx.service(); + + if (srvc != null) + return (T)srvc; + } + + return null; + } + } + finally { + leaveBusy(); + } + } + + /** {@inheritDoc} */ + @Override public ServiceContextImpl serviceContext(String name) { + if (!enterBusy()) + return null; + + try { + Collection<ServiceContextImpl> ctxs = serviceContexts(name); + + if (ctxs == null) + return null; + + synchronized (ctxs) { + if (F.isEmpty(ctxs)) + return null; + + for (ServiceContextImpl ctx : ctxs) { + if (ctx.service() != null) + return ctx; + } + } + + return null; + } + finally { + leaveBusy(); + } + } + + /** + * @param name Service name. + * @return Collection of locally deployed instance if present. + */ + @Nullable private Collection<ServiceContextImpl> serviceContexts(String name) { + IgniteUuid srvcId = lookupDeployedServiceId(name); + + if (srvcId == null) + return null; + + return locServices.get(srvcId); + } + + /** {@inheritDoc} */ + @Override public <T> T serviceProxy(ClusterGroup prj, String name, Class<? super T> srvcCls, boolean sticky, + long timeout) + throws IgniteException { + ctx.security().authorize(name, SecurityPermission.SERVICE_INVOKE, null); + + if (hasLocalNode(prj)) { + ServiceContextImpl ctx = serviceContext(name); + + if (ctx != null) { + Service srvc = ctx.service(); + + if (srvc != null) { + if (!srvcCls.isAssignableFrom(srvc.getClass())) + throw new IgniteException("Service does not implement specified interface [srvcCls=" + + srvcCls.getName() + ", srvcCls=" + srvc.getClass().getName() + ']'); + + return (T)srvc; + } + } + } + + return new GridServiceProxy<T>(prj, name, srvcCls, sticky, timeout, ctx).proxy(); + } + + /** + * @param prj Grid nodes projection. + * @return Whether given projection contains any local node. + */ + private boolean hasLocalNode(ClusterGroup prj) { + for (ClusterNode n : prj.nodes()) { + if (n.isLocal()) + return true; + } + + return false; + } + + /** {@inheritDoc} */ + @Override public <T> Collection<T> services(String name) { + if (!enterBusy()) + return null; + + try { + ctx.security().authorize(name, SecurityPermission.SERVICE_INVOKE, null); + + Collection<ServiceContextImpl> ctxs = serviceContexts(name); + + if (ctxs == null) + return null; + + synchronized (ctxs) { + if (F.isEmpty(ctxs)) + return null; + + Collection<T> res = new ArrayList<>(ctxs.size()); + + for (ServiceContextImpl ctx : ctxs) { + Service srvc = ctx.service(); + + if (srvc != null) + res.add((T)srvc); + } + + return res; + } + } + finally { + leaveBusy(); + } + } + + /** + * Reassigns service to nodes. + * + * @param srvcId Service id. + * @param cfg Service configuration. + * @param topVer Topology version. + * @param oldTop Previous topology snapshot. Will be ignored for affinity service. + * @throws IgniteCheckedException If failed. + */ + Map<UUID, Integer> reassign(@NotNull IgniteUuid srvcId, @NotNull ServiceConfiguration cfg, + @NotNull AffinityTopologyVersion topVer, + @Nullable TreeMap<UUID, Integer> oldTop) throws IgniteCheckedException { + Object nodeFilter = cfg.getNodeFilter(); + + if (nodeFilter != null) + ctx.resource().injectGeneric(nodeFilter); + + int totalCnt = cfg.getTotalCount(); + int maxPerNodeCnt = cfg.getMaxPerNodeCount(); + String cacheName = cfg.getCacheName(); + Object affKey = cfg.getAffinityKey(); + + Map<UUID, Integer> cnts = new TreeMap<>(); + + if (affKey != null && cacheName != null) { // Affinity service + ClusterNode n = ctx.affinity().mapKeyToNode(cacheName, affKey, topVer); + + if (n != null) { + int cnt = maxPerNodeCnt == 0 ? totalCnt == 0 ? 1 : totalCnt : maxPerNodeCnt; + + cnts.put(n.id(), cnt); + } + } + else { + Collection<ClusterNode> nodes = ctx.discovery().nodes(topVer); + + if (cfg.getNodeFilter() != null) { + Collection<ClusterNode> nodes0 = new ArrayList<>(); + + for (ClusterNode node : nodes) { + if (cfg.getNodeFilter().apply(node)) + nodes0.add(node); + } + + nodes = nodes0; + } + + if (!nodes.isEmpty()) { + int size = nodes.size(); + + int perNodeCnt = totalCnt != 0 ? totalCnt / size : maxPerNodeCnt; + int remainder = totalCnt != 0 ? totalCnt % size : 0; + + if (perNodeCnt >= maxPerNodeCnt && maxPerNodeCnt != 0) { + perNodeCnt = maxPerNodeCnt; + remainder = 0; + } + + for (ClusterNode n : nodes) + cnts.put(n.id(), perNodeCnt); + + assert perNodeCnt >= 0; + assert remainder >= 0; + + if (remainder > 0) { + int cnt = perNodeCnt + 1; + + Random rnd = new Random(srvcId.localId()); + + if (oldTop != null && !oldTop.isEmpty()) { + Collection<UUID> used = new TreeSet<>(); + + // Avoid redundant moving of services. + for (Map.Entry<UUID, Integer> e : oldTop.entrySet()) { + // If old count and new count match, then reuse the assignment. + if (e.getValue() == cnt) { + cnts.put(e.getKey(), cnt); + + used.add(e.getKey()); + + if (--remainder == 0) + break; + } + } + + if (remainder > 0) { + List<Map.Entry<UUID, Integer>> entries = new ArrayList<>(cnts.entrySet()); + + // Randomize. + Collections.shuffle(entries, rnd); + + for (Map.Entry<UUID, Integer> e : entries) { + // Assign only the ones that have not been reused from previous assignments. + if (!used.contains(e.getKey())) { + if (e.getValue() < maxPerNodeCnt || maxPerNodeCnt == 0) { + e.setValue(e.getValue() + 1); + + if (--remainder == 0) + break; + } + } + } + } + } + else { + List<Map.Entry<UUID, Integer>> entries = new ArrayList<>(cnts.entrySet()); + + // Randomize. + Collections.shuffle(entries, rnd); + + for (Map.Entry<UUID, Integer> e : entries) { + e.setValue(e.getValue() + 1); + + if (--remainder == 0) + break; + } + } + } + } + } + + return cnts; + } + + /** + * Redeploys local services based on assignments. + * <p/> + * Invokes from services deployment worker. + * + * @param srvcId Service id. + * @param cfg Service configuration. + * @param top Service topology. + * @throws IgniteCheckedException In case of deployment errors. + */ + void redeploy(IgniteUuid srvcId, ServiceConfiguration cfg, + Map<UUID, Integer> top) throws IgniteCheckedException { + String name = cfg.getName(); + String cacheName = cfg.getCacheName(); + Object affKey = cfg.getAffinityKey(); + + int assignCnt = top.getOrDefault(ctx.localNodeId(), 0); + + Collection<ServiceContextImpl> ctxs = locServices.computeIfAbsent(srvcId, c -> new ArrayList<>()); + + Collection<ServiceContextImpl> toInit = new ArrayList<>(); + + synchronized (ctxs) { + if (ctxs.size() > assignCnt) { + int cancelCnt = ctxs.size() - assignCnt; + + cancel(ctxs, cancelCnt); + } + else if (ctxs.size() < assignCnt) { + int createCnt = assignCnt - ctxs.size(); + + for (int i = 0; i < createCnt; i++) { + ServiceContextImpl srvcCtx = new ServiceContextImpl(name, + UUID.randomUUID(), + cacheName, + affKey, + Executors.newSingleThreadExecutor(threadFactory)); + + ctxs.add(srvcCtx); + + toInit.add(srvcCtx); + } + } + } + + for (final ServiceContextImpl srvcCtx : toInit) { + final Service srvc; + + try { + srvc = copyAndInject(cfg); + + // Initialize service. + srvc.init(srvcCtx); + + srvcCtx.service(srvc); + } + catch (Throwable e) { + U.error(log, "Failed to initialize service (service will not be deployed): " + name, e); + + synchronized (ctxs) { + ctxs.removeAll(toInit); + } + + throw new IgniteCheckedException("Error occured during service initialization: " + + "[locId=" + ctx.localNodeId() + ", name=" + name + ']', e); + } + + if (log.isInfoEnabled()) + log.info("Starting service instance [name=" + srvcCtx.name() + ", execId=" + + srvcCtx.executionId() + ']'); + + // Start service in its own thread. + final ExecutorService exe = srvcCtx.executor(); + + exe.execute(new Runnable() { + @Override public void run() { + try { + srvc.execute(srvcCtx); + } + catch (InterruptedException | IgniteInterruptedCheckedException ignore) { + if (log.isDebugEnabled()) + log.debug("Service thread was interrupted [name=" + srvcCtx.name() + ", execId=" + + srvcCtx.executionId() + ']'); + } + catch (IgniteException e) { + if (e.hasCause(InterruptedException.class) || + e.hasCause(IgniteInterruptedCheckedException.class)) { + if (log.isDebugEnabled()) + log.debug("Service thread was interrupted [name=" + srvcCtx.name() + + ", execId=" + srvcCtx.executionId() + ']'); + } + else { + U.error(log, "Service execution stopped with error [name=" + srvcCtx.name() + + ", execId=" + srvcCtx.executionId() + ']', e); + } + } + catch (Throwable e) { + U.error(log, "Service execution stopped with error [name=" + srvcCtx.name() + + ", execId=" + srvcCtx.executionId() + ']', e); + + if (e instanceof Error) + throw (Error)e; + } + finally { + // Suicide. + exe.shutdownNow(); + } + } + }); + } + } + + /** + * @param cfg Service configuration. + * @return Copy of service. + * @throws IgniteCheckedException If failed. + */ + @SuppressWarnings("deprecation") + private Service copyAndInject(ServiceConfiguration cfg) throws IgniteCheckedException { + if (cfg instanceof LazyServiceConfiguration) { + byte[] bytes = ((LazyServiceConfiguration)cfg).serviceBytes(); + + Service srvc = U.unmarshal(marsh, bytes, U.resolveClassLoader(null, ctx.config())); + + ctx.resource().inject(srvc); + + return srvc; + } + else { + Service srvc = cfg.getService(); + + try { + byte[] bytes = U.marshal(marsh, srvc); + + Service cp = U.unmarshal(marsh, bytes, U.resolveClassLoader(srvc.getClass().getClassLoader(), ctx.config())); + + ctx.resource().inject(cp); + + return cp; + } + catch (IgniteCheckedException e) { + U.error(log, "Failed to copy service (will reuse same instance): " + srvc.getClass(), e); + + return srvc; + } + } + } + + /** + * @param ctxs Contexts to cancel. + * @param cancelCnt Number of contexts to cancel. + */ + private void cancel(Iterable<ServiceContextImpl> ctxs, int cancelCnt) { + for (Iterator<ServiceContextImpl> it = ctxs.iterator(); it.hasNext(); ) { + cancel(it.next()); + + it.remove(); + + if (--cancelCnt == 0) + break; + } + } + + /** + * Perform cancelation on given service context. + * + * @param ctx Service context. + */ + private void cancel(ServiceContextImpl ctx) { + // Flip cancelled flag. + ctx.setCancelled(true); + + // Notify service about cancellation. + Service srvc = ctx.service(); + + if (srvc != null) { + try { + srvc.cancel(ctx); + } + catch (Throwable e) { + U.error(log, "Failed to cancel service (ignoring) [name=" + ctx.name() + + ", execId=" + ctx.executionId() + ']', e); + + if (e instanceof Error) + throw e; + } + finally { + try { + this.ctx.resource().cleanup(srvc); + } + catch (IgniteCheckedException e) { + U.error(log, "Failed to clean up service (will ignore): " + ctx.name(), e); + } + } + } + + // Close out executor thread for the service. + // This will cause the thread to be interrupted. + ctx.executor().shutdownNow(); + + if (log.isInfoEnabled()) { + log.info("Cancelled service instance [name=" + ctx.name() + ", execId=" + + ctx.executionId() + ']'); + } + } + + /** + * Undeployes service with given id. + * <p/> + * Invokes from services deployment worker. + * + * @param srvcId Service id. + */ + void undeploy(@NotNull IgniteUuid srvcId) { + Collection<ServiceContextImpl> ctxs = locServices.remove(srvcId); + + if (ctxs != null) { + synchronized (ctxs) { + cancel(ctxs, ctxs.size()); + } + } + } + + /** + * @param deploy {@code true} if complete deployment requests, otherwise complete undeployment request will be + * completed. + * @param reqSrvcId Request's service id. + * @param err Error to complete with. If {@code null} a future will be completed successfully. + */ + void completeInitiatingFuture(boolean deploy, IgniteUuid reqSrvcId, Throwable err) { + GridFutureAdapter<?> fut = deploy ? depFuts.remove(reqSrvcId) : undepFuts.remove(reqSrvcId); + + if (fut == null) + return; + + if (err != null) { + fut.onDone(err); + + if (deploy) { + U.warn(log, "Failed to deploy service, cfg=" + + ((GridServiceDeploymentFuture)fut).configuration(), err); + } + else + U.warn(log, "Failed to undeploy service, srvcId=" + reqSrvcId, err); + } + else + fut.onDone(); + } + + /** + * Processes deployment result. + * + * @param fullTops Deployment topologies. + */ + void updateServicesTopologies(@NotNull final Map<IgniteUuid, Map<UUID, Integer>> fullTops) { + if (!enterBusy()) + return; + + try { + updateServicesMap(deployedServices, fullTops); + } + finally { + leaveBusy(); + } + } + + /** + * @param name Service name; + * @return @return Service's id if exists, otherwise {@code null}; + */ + @Nullable private IgniteUuid lookupDeployedServiceId(String name) { + for (ServiceInfo desc : deployedServices.values()) { + if (desc.name().equals(name)) + return desc.serviceId(); + } + + return null; + } + + /** + * @param srvcId Service id. + * @return Count of locally deployed service with given id. + */ + int localInstancesCount(IgniteUuid srvcId) { + Collection<ServiceContextImpl> ctxs = locServices.get(srvcId); + + if (ctxs == null) + return 0; + + synchronized (ctxs) { + return ctxs.size(); + } + } + + /** + * Updates deployed services map according to deployment task. + * <p/> + * Invokes from services deployment worker. + * + * @param depActions Service deployment actions. + */ + void updateDeployedServices(final ServiceDeploymentActions depActions) { + if (!enterBusy()) + return; + + try { + depActions.servicesToDeploy().forEach(deployedServices::putIfAbsent); + + depActions.servicesToUndeploy().forEach((srvcId, desc) -> { + ServiceInfo rmv = deployedServices.remove(srvcId); + + assert rmv != null && rmv == desc : "Concurrent map modification."; + }); + } + finally { + leaveBusy(); + } + } + + /** + * @return Deployed services information. + */ + Map<IgniteUuid, ServiceInfo> deployedServices() { + return new HashMap<>(deployedServices); + } + + /** + * Gets services received to deploy from node with given id on joining. + * + * @param nodeId Joined node id. + * @return Services to deploy. + */ + @NotNull Map<IgniteUuid, ServiceInfo> servicesReceivedFromJoin(UUID nodeId) { + Map<IgniteUuid, ServiceInfo> descs = new HashMap<>(); + + registeredServices.forEach((srvcId, desc) -> { + if (desc.staticallyConfigured() && desc.originNodeId().equals(nodeId)) + descs.put(srvcId, desc); + }); + + return descs; + } + + /** + * @return Cluster coordinator, {@code null} if failed to determine. + */ + @Nullable ClusterNode coordinator() { + return U.oldest(ctx.discovery().aliveServerNodes(), null); + } + + /** + * @return {@code true} if local node is coordinator. + */ + private boolean isLocalNodeCoordinator() { + DiscoverySpi spi = ctx.discovery().getInjectedDiscoverySpi(); + + return spi instanceof TcpDiscoverySpi ? + ((TcpDiscoverySpi)spi).isLocalNodeCoordinator() : + F.eq(ctx.discovery().localNode(), coordinator()); + } + + /** {@inheritDoc} */ + @Override public void onLocalJoin(DiscoveryEvent evt, DiscoCache discoCache) { + assert ctx.localNodeId().equals(evt.eventNode().id()); + assert evt.type() == EVT_NODE_JOINED; + + if (isLocalNodeCoordinator()) { + // First node start, method onGridDataReceived(DiscoveryDataBag.GridDiscoveryData) has not been called. + ArrayList<ServiceInfo> staticServicesInfo = staticallyConfiguredServices(false); + + staticServicesInfo.forEach(desc -> registeredServices.put(desc.serviceId(), desc)); + } + + ServiceDeploymentActions depActions = null; + + if (!registeredServices.isEmpty()) { + depActions = new ServiceDeploymentActions(); + + depActions.servicesToDeploy(new HashMap<>(registeredServices)); + } + + depMgr.onLocalJoin(evt, discoCache, depActions); + } + + /** + * @return Services deployment manager. + */ + public ServiceDeploymentManager deployment() { + return depMgr; + } + + /** + * @param logErrors Whenever it's necessary to log validation failures. + * @return Statically configured services. + */ + @NotNull private ArrayList<ServiceInfo> staticallyConfiguredServices(boolean logErrors) { + ServiceConfiguration[] cfgs = ctx.config().getServiceConfiguration(); + + ArrayList<ServiceInfo> staticServicesInfo = new ArrayList<>(); + + if (cfgs != null) { + PreparedConfigurations<IgniteUuid> prepCfgs = prepareServiceConfigurations(Arrays.asList(cfgs), + node -> !node.isClient()); + + if (logErrors) { + if (prepCfgs.failedFuts != null) { + for (GridServiceDeploymentFuture<IgniteUuid> fut : prepCfgs.failedFuts) { + U.warn(log, "Failed to validate static service configuration (won't be deployed), " + + "cfg=" + fut.configuration() + ", err=" + fut.result()); + } + } + } + + for (ServiceConfiguration srvcCfg : prepCfgs.cfgs) + staticServicesInfo.add(new ServiceInfo(ctx.localNodeId(), IgniteUuid.randomUuid(), srvcCfg, true)); + } + + return staticServicesInfo; + } + + /** + * @param snd Sender. + * @param msg Message. + */ + private void processServicesChangeRequest(ClusterNode snd, ServiceChangeBatchRequest msg) { + DiscoveryDataClusterState state = ctx.state().clusterState(); + + if (!state.active() || state.transition()) { + for (ServiceChangeAbstractRequest req : msg.requests()) { + GridFutureAdapter<?> fut = null; + + if (req instanceof ServiceDeploymentRequest) + fut = depFuts.remove(req.serviceId()); + else if (req instanceof ServiceUndeploymentRequest) + fut = undepFuts.remove(req.serviceId()); + + if (fut != null) { + fut.onDone(new IgniteCheckedException("Operation has been canceled, cluster state " + + "change is in progress.")); + } + } + + return; + } + + Map<IgniteUuid, ServiceInfo> toDeploy = new HashMap<>(); + Map<IgniteUuid, ServiceInfo> toUndeploy = new HashMap<>(); + + for (ServiceChangeAbstractRequest req : msg.requests()) { + IgniteUuid reqSrvcId = req.serviceId(); + ServiceInfo oldDesc = registeredServices.get(reqSrvcId); + + if (req instanceof ServiceDeploymentRequest) { + IgniteCheckedException err = null; + + if (oldDesc != null) { // In case of a collision of IgniteUuid.randomUuid() (almost impossible case) + err = new IgniteCheckedException("Failed to deploy service. Service with generated id already" + + "exists : [" + "srvcId" + reqSrvcId + ", srvcTop=" + oldDesc.topologySnapshot() + ']'); + } + else { + ServiceConfiguration cfg = ((ServiceDeploymentRequest)req).configuration(); + + oldDesc = lookupInRegisteredServices(cfg.getName()); + + if (oldDesc == null) { + if (cfg.getCacheName() != null && ctx.cache().cacheDescriptor(cfg.getCacheName()) == null) { + err = new IgniteCheckedException("Failed to deploy service, " + + "affinity cache is not found, cfg=" + cfg); + } + else { + ServiceInfo desc = new ServiceInfo(snd.id(), reqSrvcId, cfg); + + registeredServices.put(reqSrvcId, desc); + + toDeploy.put(reqSrvcId, desc); + } + } + else { + if (!oldDesc.configuration().equalsIgnoreNodeFilter(cfg)) { + err = new IgniteCheckedException("Failed to deploy service " + + "(service already exists with different configuration) : " + + "[deployed=" + oldDesc.configuration() + ", new=" + cfg + ']'); + } + else { + GridServiceDeploymentFuture<IgniteUuid> fut = depFuts.remove(reqSrvcId); + + if (fut != null) { + fut.onDone(); + + if (log.isDebugEnabled()) { + log.debug("Service sent to deploy is already deployed : " + + "[srvcId=" + oldDesc.serviceId() + ", cfg=" + oldDesc.configuration()); + } + } + } + } + } + + if (err != null) { + completeInitiatingFuture(true, reqSrvcId, err); + + U.warn(log, err.getMessage(), err); + } + } + else if (req instanceof ServiceUndeploymentRequest) { + ServiceInfo rmv = registeredServices.remove(reqSrvcId); + + assert oldDesc == rmv : "Concurrent map modification."; + + toUndeploy.put(reqSrvcId, rmv); + } + } + + if (!toDeploy.isEmpty() || !toUndeploy.isEmpty()) { + ServiceDeploymentActions depActions = new ServiceDeploymentActions(); + + if (!toDeploy.isEmpty()) + depActions.servicesToDeploy(toDeploy); + + if (!toUndeploy.isEmpty()) + depActions.servicesToUndeploy(toUndeploy); + + msg.servicesDeploymentActions(depActions); + } + } + + /** + * @param msg Message. + */ + private void processChangeGlobalStateRequest(ChangeGlobalStateMessage msg) { + if (msg.activate() && registeredServices.isEmpty()) + return; + + ServiceDeploymentActions depActions = new ServiceDeploymentActions(); + + if (msg.activate()) + depActions.servicesToDeploy(new HashMap<>(registeredServices)); + else + depActions.deactivate(true); + + msg.servicesDeploymentActions(depActions); + } + + /** + * @param msg Message. + */ + private void processDynamicCacheChangeRequest(DynamicCacheChangeBatch msg) { + Map<IgniteUuid, ServiceInfo> toUndeploy = new HashMap<>(); + + for (DynamicCacheChangeRequest chReq : msg.requests()) { + if (chReq.stop()) { + registeredServices.entrySet().removeIf(e -> { + ServiceInfo desc = e.getValue(); + + if (desc.cacheName().equals(chReq.cacheName())) { + toUndeploy.put(desc.serviceId(), desc); + + return true; + } + + return false; + }); + } + } + + if (!toUndeploy.isEmpty()) { + ServiceDeploymentActions depActions = new ServiceDeploymentActions(); + + depActions.servicesToUndeploy(toUndeploy); + + msg.servicesDeploymentActions(depActions); + } + } + + /** + * @param msg Message. + */ + private void processServicesFullDeployments(ServiceClusterDeploymentResultBatch msg) { + final Map<IgniteUuid, Map<UUID, Integer>> fullTops = new HashMap<>(); + final Map<IgniteUuid, Collection<byte[]>> fullErrors = new HashMap<>(); + + for (ServiceClusterDeploymentResult depRes : msg.results()) { + final IgniteUuid srvcId = depRes.serviceId(); + final Map<UUID, ServiceSingleNodeDeploymentResult> deps = depRes.results(); + + final Map<UUID, Integer> top = new HashMap<>(); + final Collection<byte[]> errors = new ArrayList<>(); + + deps.forEach((nodeId, res) -> { + int cnt = res.count(); + + if (cnt > 0) + top.put(nodeId, cnt); + + if (!res.errors().isEmpty()) + errors.addAll(res.errors()); + }); + + if (!errors.isEmpty()) + fullErrors.computeIfAbsent(srvcId, e -> new ArrayList<>()).addAll(errors); + + fullTops.put(srvcId, top); + } + + synchronized (servicesTopsUpdateMux) { + updateServicesMap(registeredServices, fullTops); + + servicesTopsUpdateMux.notifyAll(); + } + + ServiceDeploymentActions depActions = new ServiceDeploymentActions(); + + depActions.deploymentTopologies(fullTops); + depActions.deploymentErrors(fullErrors); + + msg.servicesDeploymentActions(depActions); + } + + /** + * @param name Service name. + * @return Mapped service descriptor. Possibly {@code null} if not found. + */ + @Nullable private ServiceInfo lookupInRegisteredServices(String name) { + for (ServiceInfo desc : registeredServices.values()) { + if (desc.name().equals(name)) + return desc; + } + + return null; + } + + /** + * Updates services info according to given arguments. + * + * @param services Services info to update. + * @param tops Deployment topologies. + */ + private void updateServicesMap(Map<IgniteUuid, ServiceInfo> services, + Map<IgniteUuid, Map<UUID, Integer>> tops) { + + tops.forEach((srvcId, top) -> { + ServiceInfo desc = services.get(srvcId); + + if (desc != null) + desc.topologySnapshot(top); + }); + } + + /** + * Enters busy state. + * + * @return {@code true} if entered to busy state. + */ + private boolean enterBusy() { + return opsLock.readLock().tryLock(); + } + + /** + * Leaves busy state. + */ + private void leaveBusy() { + opsLock.readLock().unlock(); + } +}
http://git-wip-us.apache.org/repos/asf/ignite/blob/62c560a5/modules/core/src/main/java/org/apache/ignite/internal/processors/service/PreparedConfigurations.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/service/PreparedConfigurations.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/service/PreparedConfigurations.java index dc41c22..abfb65c 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/service/PreparedConfigurations.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/service/PreparedConfigurations.java @@ -17,6 +17,7 @@ package org.apache.ignite.internal.processors.service; +import java.io.Serializable; import java.util.List; import org.apache.ignite.internal.util.typedef.internal.S; import org.apache.ignite.services.ServiceConfiguration; @@ -24,18 +25,18 @@ import org.apache.ignite.services.ServiceConfiguration; /** * Result of services validation before deployment. */ -class PreparedConfigurations { +class PreparedConfigurations<T extends Serializable> { /** */ final List<ServiceConfiguration> cfgs; /** */ - final List<GridServiceDeploymentFuture> failedFuts; + final List<GridServiceDeploymentFuture<T>> failedFuts; /** * @param cfgs Configurations to deploy. * @param failedFuts Finished futures for failed configurations. */ - PreparedConfigurations(List<ServiceConfiguration> cfgs, List<GridServiceDeploymentFuture> failedFuts) { + PreparedConfigurations(List<ServiceConfiguration> cfgs, List<GridServiceDeploymentFuture<T>> failedFuts) { this.cfgs = cfgs; this.failedFuts = failedFuts; } http://git-wip-us.apache.org/repos/asf/ignite/blob/62c560a5/modules/core/src/main/java/org/apache/ignite/internal/processors/service/ServiceChangeAbstractRequest.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/service/ServiceChangeAbstractRequest.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/service/ServiceChangeAbstractRequest.java new file mode 100644 index 0000000..d083aad --- /dev/null +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/service/ServiceChangeAbstractRequest.java @@ -0,0 +1,53 @@ +/* + * 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.processors.service; + +import java.io.Serializable; +import org.apache.ignite.internal.util.typedef.internal.S; +import org.apache.ignite.lang.IgniteUuid; +import org.jetbrains.annotations.NotNull; + +/** + * Abstract class for service change requests. + */ +abstract class ServiceChangeAbstractRequest implements Serializable { + /** */ + private static final long serialVersionUID = 0L; + + /** Service id. */ + protected final IgniteUuid srvcId; + + /** + * @param srvcId Service id. + */ + protected ServiceChangeAbstractRequest(@NotNull IgniteUuid srvcId) { + this.srvcId = srvcId; + } + + /** + * @return Service id. + */ + public IgniteUuid serviceId() { + return srvcId; + } + + /** {@inheritDoc} */ + @Override public String toString() { + return S.toString(ServiceChangeAbstractRequest.class, this); + } +} http://git-wip-us.apache.org/repos/asf/ignite/blob/62c560a5/modules/core/src/main/java/org/apache/ignite/internal/processors/service/ServiceChangeBatchRequest.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/service/ServiceChangeBatchRequest.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/service/ServiceChangeBatchRequest.java new file mode 100644 index 0000000..a6707b6 --- /dev/null +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/service/ServiceChangeBatchRequest.java @@ -0,0 +1,113 @@ +/* + * 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.processors.service; + +import java.util.Collection; +import java.util.Collections; +import org.apache.ignite.internal.managers.discovery.DiscoCache; +import org.apache.ignite.internal.managers.discovery.DiscoveryCustomMessage; +import org.apache.ignite.internal.managers.discovery.GridDiscoveryManager; +import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion; +import org.apache.ignite.internal.util.tostring.GridToStringExclude; +import org.apache.ignite.internal.util.tostring.GridToStringInclude; +import org.apache.ignite.internal.util.typedef.F; +import org.apache.ignite.internal.util.typedef.internal.S; +import org.apache.ignite.lang.IgniteUuid; +import org.jetbrains.annotations.Nullable; + +/** + * Service change batch request discovery message. + */ +public class ServiceChangeBatchRequest implements DiscoveryCustomMessage { + /** */ + private static final long serialVersionUID = 0L; + + /** Unique custom message ID. */ + private final IgniteUuid id = IgniteUuid.randomUuid(); + + /** Change requests. */ + @GridToStringInclude + private final Collection<ServiceChangeAbstractRequest> reqs; + + /** Services deployment actions to be processed on services deployment process. */ + @GridToStringExclude + @Nullable private transient ServiceDeploymentActions serviceDeploymentActions; + + /** + * @param reqs Change requests. + */ + public ServiceChangeBatchRequest(Collection<ServiceChangeAbstractRequest> reqs) { + assert !F.isEmpty(reqs); + + this.reqs = reqs; + } + + /** + * @return Change requests. + */ + public Collection<ServiceChangeAbstractRequest> requests() { + return Collections.unmodifiableCollection(reqs); + } + + /** + * @return Services deployment actions to be processed on services deployment process. + */ + @Nullable public ServiceDeploymentActions servicesDeploymentActions() { + return serviceDeploymentActions; + } + + /** + * @param serviceDeploymentActions Services deployment actions to be processed on services deployment process. + */ + public void servicesDeploymentActions(ServiceDeploymentActions serviceDeploymentActions) { + this.serviceDeploymentActions = serviceDeploymentActions; + } + + /** {@inheritDoc} */ + @Override public IgniteUuid id() { + return id; + } + + /** {@inheritDoc} */ + @Nullable @Override public DiscoveryCustomMessage ackMessage() { + // No-op. + return null; + } + + /** {@inheritDoc} */ + @Override public boolean isMutable() { + return false; + } + + /** {@inheritDoc} */ + @Override public boolean stopProcess() { + return false; + } + + /** {@inheritDoc} */ + @Override public DiscoCache createDiscoCache(GridDiscoveryManager mgr, AffinityTopologyVersion topVer, + DiscoCache discoCache) { + // No-op. + return null; + } + + /** {@inheritDoc} */ + @Override public String toString() { + return S.toString(ServiceChangeBatchRequest.class, this); + } +} http://git-wip-us.apache.org/repos/asf/ignite/blob/62c560a5/modules/core/src/main/java/org/apache/ignite/internal/processors/service/ServiceClusterDeploymentResult.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/service/ServiceClusterDeploymentResult.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/service/ServiceClusterDeploymentResult.java new file mode 100644 index 0000000..45e2c52 --- /dev/null +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/service/ServiceClusterDeploymentResult.java @@ -0,0 +1,73 @@ +/* + * 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.processors.service; + +import java.io.Serializable; +import java.util.Collections; +import java.util.Map; +import java.util.UUID; +import org.apache.ignite.internal.util.tostring.GridToStringInclude; +import org.apache.ignite.internal.util.typedef.internal.S; +import org.apache.ignite.lang.IgniteUuid; +import org.jetbrains.annotations.NotNull; + +/** + * Service cluster deployment result. + * <p/> + * Contains coint of deployed service and deployment errors across the cluster mapped to nodes ids. + */ +public class ServiceClusterDeploymentResult implements Serializable { + /** */ + private static final long serialVersionUID = 0L; + + /** Service id. */ + private final IgniteUuid srvcId; + + /** Per node deployments results. */ + @GridToStringInclude + private final Map<UUID, ServiceSingleNodeDeploymentResult> results; + + /** + * @param srvcId Service id. + * @param results Deployments results. + */ + public ServiceClusterDeploymentResult(@NotNull IgniteUuid srvcId, + @NotNull Map<UUID, ServiceSingleNodeDeploymentResult> results) { + this.srvcId = srvcId; + this.results = results; + } + + /** + * @return Service id. + */ + public IgniteUuid serviceId() { + return srvcId; + } + + /** + * @return Per node deployments results. + */ + public Map<UUID, ServiceSingleNodeDeploymentResult> results() { + return Collections.unmodifiableMap(results); + } + + /** {@inheritDoc} */ + @Override public String toString() { + return S.toString(ServiceClusterDeploymentResult.class, this); + } +} http://git-wip-us.apache.org/repos/asf/ignite/blob/62c560a5/modules/core/src/main/java/org/apache/ignite/internal/processors/service/ServiceClusterDeploymentResultBatch.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/service/ServiceClusterDeploymentResultBatch.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/service/ServiceClusterDeploymentResultBatch.java new file mode 100644 index 0000000..6aeb4b6 --- /dev/null +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/service/ServiceClusterDeploymentResultBatch.java @@ -0,0 +1,125 @@ +/* + * 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.processors.service; + +import java.util.Collection; +import org.apache.ignite.internal.managers.discovery.DiscoCache; +import org.apache.ignite.internal.managers.discovery.DiscoveryCustomMessage; +import org.apache.ignite.internal.managers.discovery.GridDiscoveryManager; +import org.apache.ignite.internal.processors.affinity.AffinityTopologyVersion; +import org.apache.ignite.internal.util.tostring.GridToStringExclude; +import org.apache.ignite.internal.util.tostring.GridToStringInclude; +import org.apache.ignite.internal.util.typedef.internal.S; +import org.apache.ignite.lang.IgniteUuid; +import org.jetbrains.annotations.NotNull; +import org.jetbrains.annotations.Nullable; + +/** + * Batch of services cluster deployment result. + * <p/> + * Contains collection of {@link ServiceClusterDeploymentResult}. + */ +public class ServiceClusterDeploymentResultBatch implements DiscoveryCustomMessage { + /** */ + private static final long serialVersionUID = 0L; + + /** Unique custom message ID. */ + private final IgniteUuid id = IgniteUuid.randomUuid(); + + /** Deployment process id. */ + private final ServiceDeploymentProcessId depId; + + /** Services deployments results. */ + @GridToStringInclude + private Collection<ServiceClusterDeploymentResult> results; + + /** Services deployment actions to be processed on services deployment process. */ + @GridToStringExclude + @Nullable private transient ServiceDeploymentActions serviceDeploymentActions; + + /** + * @param depId Deployment process id. + * @param results Services deployments results. + */ + public ServiceClusterDeploymentResultBatch(@NotNull ServiceDeploymentProcessId depId, + @NotNull Collection<ServiceClusterDeploymentResult> results) { + this.depId = depId; + this.results = results; + } + + /** + * @return Deployment process id. + */ + public ServiceDeploymentProcessId deploymentId() { + return depId; + } + + /** + * @return Services deployments results. + */ + public Collection<ServiceClusterDeploymentResult> results() { + return results; + } + + /** + * @return Services deployment actions to be processed on services deployment process. + */ + @Nullable public ServiceDeploymentActions servicesDeploymentActions() { + return serviceDeploymentActions; + } + + /** + * @param serviceDeploymentActions Services deployment actions to be processed on services deployment process. + */ + public void servicesDeploymentActions(ServiceDeploymentActions serviceDeploymentActions) { + this.serviceDeploymentActions = serviceDeploymentActions; + } + + /** {@inheritDoc} */ + @Override public IgniteUuid id() { + return id; + } + + /** {@inheritDoc} */ + @Nullable @Override public DiscoveryCustomMessage ackMessage() { + // No-op. + return null; + } + + /** {@inheritDoc} */ + @Override public boolean isMutable() { + return false; + } + + /** {@inheritDoc} */ + @Override public boolean stopProcess() { + return false; + } + + /** {@inheritDoc} */ + @Override public DiscoCache createDiscoCache(GridDiscoveryManager mgr, AffinityTopologyVersion topVer, + DiscoCache discoCache) { + // No-op. + return null; + } + + /** {@inheritDoc} */ + @Override public String toString() { + return S.toString(ServiceClusterDeploymentResultBatch.class, this); + } +} http://git-wip-us.apache.org/repos/asf/ignite/blob/62c560a5/modules/core/src/main/java/org/apache/ignite/internal/processors/service/ServiceDeploymentActions.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/service/ServiceDeploymentActions.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/service/ServiceDeploymentActions.java new file mode 100644 index 0000000..b33a646 --- /dev/null +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/service/ServiceDeploymentActions.java @@ -0,0 +1,115 @@ +/* + * 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.processors.service; + +import java.util.Collection; +import java.util.Collections; +import java.util.Map; +import java.util.UUID; +import org.apache.ignite.lang.IgniteUuid; +import org.jetbrains.annotations.NotNull; + +/** + * Actions of change service state to be processed in the service deployment process. + */ +public class ServiceDeploymentActions { + /** Whenever it's necessary to deactivate service processor. */ + private boolean deactivate; + + /** Services info to deploy. */ + private Map<IgniteUuid, ServiceInfo> servicesToDeploy; + + /** Services info to undeploy. */ + private Map<IgniteUuid, ServiceInfo> servicesToUndeploy; + + /** Services deployment topologies. */ + private Map<IgniteUuid, Map<UUID, Integer>> depTops; + + /** Services deployment errors. */ + private Map<IgniteUuid, Collection<byte[]>> depErrors; + + /** + * @param servicesToDeploy Services info to deploy. + */ + public void servicesToDeploy(@NotNull Map<IgniteUuid, ServiceInfo> servicesToDeploy) { + this.servicesToDeploy = servicesToDeploy; + } + + /** + * @return Services info to deploy. + */ + @NotNull public Map<IgniteUuid, ServiceInfo> servicesToDeploy() { + return servicesToDeploy != null ? servicesToDeploy : Collections.emptyMap(); + } + + /** + * @param servicesToUndeploy Services info to undeploy. + */ + public void servicesToUndeploy(@NotNull Map<IgniteUuid, ServiceInfo> servicesToUndeploy) { + this.servicesToUndeploy = servicesToUndeploy; + } + + /** + * @return Services info to undeploy. + */ + @NotNull public Map<IgniteUuid, ServiceInfo> servicesToUndeploy() { + return servicesToUndeploy != null ? servicesToUndeploy : Collections.emptyMap(); + } + + /** + * @param deactivate Whenever it's necessary to deactivate service processor. + */ + public void deactivate(boolean deactivate) { + this.deactivate = deactivate; + } + + /** + * @return Whenever it's necessary to deactivate service processor. + */ + public boolean deactivate() { + return deactivate; + } + + /** + * @return Deployment topologies. + */ + @NotNull public Map<IgniteUuid, Map<UUID, Integer>> deploymentTopologies() { + return depTops != null ? depTops : Collections.emptyMap(); + } + + /** + * @param depTops Deployment topologies. + */ + public void deploymentTopologies(@NotNull Map<IgniteUuid, Map<UUID, Integer>> depTops) { + this.depTops = depTops; + } + + /** + * @return Deployment errors. + */ + @NotNull public Map<IgniteUuid, Collection<byte[]>> deploymentErrors() { + return depErrors != null ? depErrors : Collections.emptyMap(); + } + + /** + * @param depErrors Deployment errors. + */ + public void deploymentErrors(@NotNull Map<IgniteUuid, Collection<byte[]>> depErrors) { + this.depErrors = depErrors; + } +}
