| /* |
| * Copyright 2016-present Open Networking Foundation |
| * |
| * Licensed 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.onosproject.store.app; |
| |
| import com.google.common.base.Charsets; |
| import com.google.common.base.MoreObjects; |
| import com.google.common.base.Preconditions; |
| import com.google.common.base.Throwables; |
| import com.google.common.collect.ImmutableSet; |
| import com.google.common.collect.Lists; |
| import com.google.common.collect.Maps; |
| import com.google.common.collect.Multimap; |
| import com.google.common.collect.Sets; |
| import org.onosproject.app.ApplicationDescription; |
| import org.onosproject.app.ApplicationEvent; |
| import org.onosproject.app.ApplicationException; |
| import org.onosproject.app.ApplicationIdStore; |
| import org.onosproject.app.ApplicationState; |
| import org.onosproject.app.ApplicationStore; |
| import org.onosproject.app.ApplicationStoreDelegate; |
| import org.onosproject.cluster.ClusterService; |
| import org.onosproject.cluster.ControllerNode; |
| import org.onosproject.common.app.ApplicationArchive; |
| import org.onosproject.core.Application; |
| import org.onosproject.core.ApplicationId; |
| import org.onosproject.core.CoreService; |
| import org.onosproject.core.DefaultApplication; |
| import org.onosproject.core.Version; |
| import org.onosproject.core.VersionService; |
| import org.onosproject.security.Permission; |
| import org.onosproject.store.cluster.messaging.ClusterCommunicationService; |
| import org.onosproject.store.cluster.messaging.MessageSubject; |
| import org.onosproject.store.serializers.KryoNamespaces; |
| import org.onosproject.store.service.ConsistentMap; |
| import org.onosproject.store.service.DistributedPrimitive.Status; |
| import org.onosproject.store.service.MapEvent; |
| import org.onosproject.store.service.MapEventListener; |
| import org.onosproject.store.service.RevisionType; |
| import org.onosproject.store.service.Serializer; |
| import org.onosproject.store.service.StorageException; |
| import org.onosproject.store.service.StorageService; |
| import org.onosproject.store.service.Topic; |
| import org.onosproject.store.service.Versioned; |
| import org.osgi.service.component.annotations.Activate; |
| import org.osgi.service.component.annotations.Component; |
| import org.osgi.service.component.annotations.Deactivate; |
| import org.osgi.service.component.annotations.Reference; |
| import org.osgi.service.component.annotations.ReferenceCardinality; |
| import org.slf4j.Logger; |
| |
| import java.io.ByteArrayInputStream; |
| import java.io.IOException; |
| import java.io.InputStream; |
| import java.util.List; |
| import java.util.Optional; |
| import java.util.Set; |
| import java.util.concurrent.CountDownLatch; |
| import java.util.concurrent.ExecutorService; |
| import java.util.concurrent.ScheduledExecutorService; |
| import java.util.concurrent.TimeUnit; |
| import java.util.concurrent.atomic.AtomicBoolean; |
| import java.util.function.Consumer; |
| import java.util.function.Function; |
| import java.util.stream.Collectors; |
| |
| import static com.google.common.collect.Multimaps.newSetMultimap; |
| import static com.google.common.collect.Multimaps.synchronizedSetMultimap; |
| import static com.google.common.io.ByteStreams.toByteArray; |
| import static java.util.concurrent.Executors.newSingleThreadExecutor; |
| import static java.util.concurrent.Executors.newSingleThreadScheduledExecutor; |
| import static java.util.concurrent.TimeUnit.MILLISECONDS; |
| import static org.onlab.util.Tools.groupedThreads; |
| import static org.onlab.util.Tools.randomDelay; |
| import static org.onosproject.app.ApplicationEvent.Type.APP_ACTIVATED; |
| import static org.onosproject.app.ApplicationEvent.Type.APP_DEACTIVATED; |
| import static org.onosproject.app.ApplicationEvent.Type.APP_INSTALLED; |
| import static org.onosproject.app.ApplicationEvent.Type.APP_PERMISSIONS_CHANGED; |
| import static org.onosproject.app.ApplicationEvent.Type.APP_UNINSTALLED; |
| import static org.onosproject.store.app.DistributedApplicationStore.InternalState.ACTIVATED; |
| import static org.onosproject.store.app.DistributedApplicationStore.InternalState.DEACTIVATED; |
| import static org.onosproject.store.app.DistributedApplicationStore.InternalState.INSTALLED; |
| import static org.slf4j.LoggerFactory.getLogger; |
| |
| /** |
| * Manages inventory of applications in a distributed data store providing |
| * stronger consistency guarantees. |
| */ |
| @Component(immediate = true, service = ApplicationStore.class) |
| public class DistributedApplicationStore extends ApplicationArchive |
| implements ApplicationStore { |
| |
| private final Logger log = getLogger(getClass()); |
| |
| private static final MessageSubject APP_BITS_REQUEST = new MessageSubject("app-bits-request"); |
| |
| private static final int MAX_LOAD_RETRIES = 5; |
| private static final int RETRY_DELAY_MS = 2_000; |
| |
| private static final int FETCH_TIMEOUT_MS = 10_000; |
| |
| private static final int APP_LOAD_DELAY_MS = 500; |
| |
| private static List<String> pendingApps = Lists.newArrayList(); |
| |
| public enum InternalState { |
| INSTALLED, ACTIVATED, DEACTIVATED |
| } |
| |
| private ScheduledExecutorService executor; |
| private ExecutorService messageHandlingExecutor, activationExecutor; |
| |
| private ConsistentMap<ApplicationId, InternalApplicationHolder> apps; |
| private Topic<Application> appActivationTopic; |
| |
| @Reference(cardinality = ReferenceCardinality.MANDATORY) |
| protected ClusterCommunicationService clusterCommunicator; |
| |
| @Reference(cardinality = ReferenceCardinality.MANDATORY) |
| protected ClusterService clusterService; |
| |
| @Reference(cardinality = ReferenceCardinality.MANDATORY) |
| protected StorageService storageService; |
| |
| @Reference(cardinality = ReferenceCardinality.MANDATORY) |
| protected ApplicationIdStore idStore; |
| |
| @Reference(cardinality = ReferenceCardinality.MANDATORY) |
| protected VersionService versionService; |
| |
| private final InternalAppsListener appsListener = new InternalAppsListener(); |
| private final Consumer<Application> appActivator = new AppActivator(); |
| |
| private Consumer<Status> statusChangeListener; |
| |
| // Multimap to track which apps are required by others apps |
| // app -> { required-by, ... } |
| // Apps explicitly activated will be required by the CORE app |
| private final Multimap<ApplicationId, ApplicationId> requiredBy = |
| synchronizedSetMultimap(newSetMultimap(Maps.newHashMap(), Sets::newHashSet)); |
| |
| private ApplicationId coreAppId; |
| |
| // Apps started in this node. |
| private final Set<String> localStartedApps = Sets.newConcurrentHashSet(); |
| |
| @Activate |
| public void activate() { |
| messageHandlingExecutor = newSingleThreadExecutor(groupedThreads("onos/store/app", |
| "message-handler", log)); |
| clusterCommunicator.addSubscriber(APP_BITS_REQUEST, |
| bytes -> new String(bytes, Charsets.UTF_8), |
| name -> { |
| try { |
| log.info("Sending bits for application {}", name); |
| return toByteArray(getApplicationInputStream(name)); |
| } catch (ApplicationException e) { |
| log.warn("Bits for application {} are not available on this node yet", name); |
| return null; |
| } catch (IOException e) { |
| throw new StorageException(e); |
| } |
| }, |
| Function.identity(), |
| messageHandlingExecutor); |
| |
| apps = storageService.<ApplicationId, InternalApplicationHolder>consistentMapBuilder() |
| .withName("onos-apps") |
| .withRelaxedReadConsistency() |
| .withSerializer(Serializer.using(KryoNamespaces.API, |
| InternalApplicationHolder.class, |
| InternalState.class)) |
| .withVersion(versionService.version()) |
| .withRevisionType(RevisionType.PROPAGATE) |
| .withCompatibilityFunction(this::convertApplication) |
| .build(); |
| |
| /* To update the version in application store if it does not matche with the local system. |
| This will happen only when upgrading onos using issu or during rollback if upgrade does not work. |
| */ |
| apps.asJavaMap().forEach((appId, holder) -> apps.asJavaMap() |
| .put(appId, convertApplication(holder, versionService.version()))); |
| |
| appActivationTopic = storageService.<Application>topicBuilder() |
| .withName("onos-apps-activation-topic") |
| .withSerializer(Serializer.using(KryoNamespaces.API)) |
| .withVersion(versionService.version()) |
| .withRevisionType(RevisionType.PROPAGATE) |
| .withCompatibilityFunction(this::convertApplication) |
| .build(); |
| |
| activationExecutor = newSingleThreadExecutor(groupedThreads("onos/store/app", |
| "app-activation", log)); |
| appActivationTopic.subscribe(appActivator, activationExecutor); |
| |
| executor = newSingleThreadScheduledExecutor(groupedThreads("onos/app", "store", log)); |
| statusChangeListener = status -> { |
| if (status == Status.ACTIVE) { |
| executor.execute(this::bootstrapExistingApplications); |
| } |
| }; |
| apps.addListener(appsListener, activationExecutor); |
| apps.addStatusChangeListener(statusChangeListener); |
| coreAppId = getId(CoreService.CORE_APP_NAME); |
| |
| downloadMissingApplications(); |
| activateExistingApplications(); |
| log.info("Started"); |
| } |
| |
| /** |
| * Converts the version of the application in store to the version of the local application. |
| */ |
| private InternalApplicationHolder convertApplication(InternalApplicationHolder appHolder, Version version) { |
| // Load the application description from disk. If the version doesn't match the persisted |
| // version, update the stored application with the new version. |
| ApplicationDescription appDesc = null; |
| try { |
| appDesc = getApplicationDescription(appHolder.app.id().name()); |
| } catch (ApplicationException e) { |
| // If external application is not present then just ignore it as it will be installed from other onos nodes |
| log.warn("Application : {} not found in disk", appHolder.app.id().name()); |
| } |
| if (appDesc != null && !appDesc.version().equals(appHolder.app().version())) { |
| log.info("Updating app version to : {} in store for app : {}", appDesc.version(), appHolder.app.id()); |
| Application newApplication = DefaultApplication.builder(appDesc) |
| .withAppId(appHolder.app.id()) |
| .build(); |
| return new InternalApplicationHolder( |
| newApplication, appHolder.state, appHolder.permissions); |
| } |
| return appHolder; |
| } |
| |
| /** |
| * Converts the versions of stored applications propagated from the prior version to the local application versions. |
| */ |
| private Application convertApplication(Application app, Version version) { |
| // Load the application description from disk. If the version doesn't match the persisted |
| // version, update the stored application with the new version. |
| ApplicationDescription appDesc = getApplicationDescription(app.id().name()); |
| if (!appDesc.version().equals(app.version())) { |
| return DefaultApplication.builder(appDesc) |
| .withAppId(app.id()) |
| .build(); |
| } |
| return app; |
| } |
| |
| /** |
| * Downloads any missing bits for installed applications. |
| */ |
| private void downloadMissingApplications() { |
| log.info("Going to download missing applications"); |
| apps.asJavaMap().forEach((appId, holder) -> fetchBitsIfNeeded(holder.app)); |
| } |
| |
| /** |
| * Activates applications that should be activated according to the distributed store. |
| */ |
| private void activateExistingApplications() { |
| log.info("Going to activate existing applications"); |
| getApplicationNames().forEach(appName -> { |
| // Only update the application version if the application has already been installed. |
| ApplicationId appId = getId(appName); |
| if (appId != null) { |
| ApplicationDescription appDesc = getApplicationDescription(appName); |
| InternalApplicationHolder appHolder = Versioned.valueOrNull(apps.get(appId)); |
| |
| if (appHolder != null && appHolder.state == ACTIVATED) { |
| log.debug("App name and version from local system : {}, {}", appDesc.name(), appDesc.version()); |
| log.debug("App name and version from app store : {}, {}", appHolder.app.id(), |
| appHolder.app().version()); |
| } |
| // If the application has already been activated, set the local state to active. |
| if (appHolder != null |
| && appDesc.version().equals(appHolder.app().version()) |
| && appHolder.state == ACTIVATED) { |
| log.info("Going to activate app : {}", appHolder.app.id()); |
| setActive(appName); |
| updateTime(appName); |
| } |
| } |
| }); |
| } |
| |
| |
| /** |
| * Processes existing applications from the distributed map. This is done to |
| * account for events that this instance may be have missed due to a staggered start. |
| */ |
| private void bootstrapExistingApplications() { |
| apps.asJavaMap().forEach((appId, holder) -> setupApplicationAndNotify(appId, holder.app(), holder.state())); |
| } |
| |
| /** |
| * Loads the application inventory from the disk and activates apps if |
| * they are marked to be active. |
| */ |
| private void loadFromDisk() { |
| getApplicationNames().forEach(appName -> { |
| Application app = loadFromDisk(appName); |
| if (app != null && isActive(app.id().name())) { |
| // For now, apps loaded from disk will be marked as having been |
| // activated explicitly, which means they won't deactivate |
| // implicitly when all dependent apps have been deactivated. |
| requiredBy.put(app.id(), coreAppId); |
| activate(app.id(), false); |
| // TODO Load app permissions |
| } |
| }); |
| } |
| |
| private Application loadFromDisk(String appName) { |
| pendingApps.add(appName); |
| |
| for (int i = 0; i < MAX_LOAD_RETRIES; i++) { |
| try { |
| // Directly return if app already exists |
| ApplicationId appId = getId(appName); |
| if (appId != null) { |
| Application application = getApplication(appId); |
| if (application != null) { |
| pendingApps.remove(appName); |
| return application; |
| } |
| } |
| |
| ApplicationDescription appDesc = getApplicationDescription(appName); |
| |
| Optional<String> loop = appDesc.requiredApps().stream() |
| .filter(app -> pendingApps.contains(app)).findAny(); |
| if (loop.isPresent()) { |
| log.error("Circular app dependency detected: {} -> {}", pendingApps, loop.get()); |
| pendingApps.remove(appName); |
| return null; |
| } |
| |
| boolean success = appDesc.requiredApps().stream() |
| .noneMatch(requiredApp -> loadFromDisk(requiredApp) == null); |
| pendingApps.remove(appName); |
| |
| if (success) { |
| return create(appDesc, false); |
| } else { |
| log.error("Unable to load dependencies for application {}", appName); |
| return null; |
| } |
| |
| } catch (Exception e) { |
| log.warn("Unable to load application {} from disk: {}; retrying", |
| appName, |
| Throwables.getRootCause(e).getMessage()); |
| log.debug("Full error details:", e); |
| randomDelay(RETRY_DELAY_MS); //FIXME: This is a deliberate hack; fix in Falcon |
| } |
| } |
| pendingApps.remove(appName); |
| log.error("Unable to load application {}", appName); |
| return null; |
| } |
| |
| @Deactivate |
| public void deactivate() { |
| clusterCommunicator.removeSubscriber(APP_BITS_REQUEST); |
| apps.removeStatusChangeListener(statusChangeListener); |
| apps.removeListener(appsListener); |
| appActivationTopic.unsubscribe(appActivator); |
| messageHandlingExecutor.shutdown(); |
| activationExecutor.shutdown(); |
| executor.shutdown(); |
| log.info("Stopped"); |
| } |
| |
| @Override |
| public void setDelegate(ApplicationStoreDelegate delegate) { |
| super.setDelegate(delegate); |
| executor.execute(this::bootstrapExistingApplications); |
| executor.schedule((Runnable) this::loadFromDisk, APP_LOAD_DELAY_MS, TimeUnit.MILLISECONDS); |
| } |
| |
| @Override |
| public Set<Application> getApplications() { |
| return ImmutableSet.copyOf(apps.values() |
| .stream() |
| .map(Versioned::value) |
| .map(InternalApplicationHolder::app) |
| .collect(Collectors.toSet())); |
| } |
| |
| @Override |
| public ApplicationId getId(String name) { |
| return idStore.getAppId(name); |
| } |
| |
| @Override |
| public Application getApplication(ApplicationId appId) { |
| InternalApplicationHolder appHolder = Versioned.valueOrNull(apps.get(appId)); |
| return appHolder != null ? appHolder.app() : null; |
| } |
| |
| @Override |
| public ApplicationState getState(ApplicationId appId) { |
| InternalApplicationHolder appHolder = Versioned.valueOrNull(apps.get(appId)); |
| InternalState state = appHolder != null ? appHolder.state() : null; |
| return state == null ? null : state == ACTIVATED ? ApplicationState.ACTIVE : ApplicationState.INSTALLED; |
| } |
| |
| @Override |
| public Application create(InputStream appDescStream) { |
| ApplicationDescription appDesc = saveApplication(appDescStream); |
| if (hasPrerequisites(appDesc)) { |
| return create(appDesc, true); |
| } |
| // Purge bits off disk if we don't have prerequisites to allow app to be |
| // reinstalled later |
| purgeApplication(appDesc.name()); |
| throw new ApplicationException("Missing dependencies for app " + appDesc.name()); |
| } |
| |
| private boolean hasPrerequisites(ApplicationDescription app) { |
| for (String required : app.requiredApps()) { |
| ApplicationId id = getId(required); |
| if (id == null || getApplication(id) == null) { |
| log.error("{} required for {} not available", required, app.name()); |
| return false; |
| } |
| } |
| return true; |
| } |
| |
| private Application create(ApplicationDescription appDesc, boolean updateTime) { |
| Application app = registerApp(appDesc); |
| if (updateTime) { |
| updateTime(app.id().name()); |
| } |
| InternalApplicationHolder previousApp = |
| Versioned.valueOrNull(apps.putIfAbsent(app.id(), new InternalApplicationHolder(app, INSTALLED, null))); |
| return previousApp != null ? previousApp.app() : app; |
| } |
| |
| @Override |
| public void remove(ApplicationId appId) { |
| uninstallDependentApps(appId); |
| apps.remove(appId); |
| } |
| |
| // Uninstalls all apps that depend on the given app. |
| private void uninstallDependentApps(ApplicationId appId) { |
| getApplications().stream() |
| .filter(a -> a.requiredApps().contains(appId.name())) |
| .forEach(a -> remove(a.id())); |
| } |
| |
| @Override |
| public void activate(ApplicationId appId) { |
| activate(appId, coreAppId); |
| } |
| |
| private void activate(ApplicationId appId, ApplicationId forAppId) { |
| requiredBy.put(appId, forAppId); |
| activate(appId, true); |
| } |
| |
| private void activate(ApplicationId appId, boolean updateTime) { |
| Versioned<InternalApplicationHolder> vAppHolder = apps.get(appId); |
| if (vAppHolder != null) { |
| if (log.isTraceEnabled()) { |
| log.trace("Activating {}", appId); |
| } |
| if (updateTime) { |
| updateTime(appId.name()); |
| } |
| activateRequiredApps(vAppHolder.value().app()); |
| |
| apps.computeIf(appId, v -> v != null && v.state() != ACTIVATED, |
| (k, v) -> new InternalApplicationHolder( |
| v.app(), ACTIVATED, v.permissions())); |
| appActivationTopic.publish(vAppHolder.value().app()); |
| appActivationTopic.publish(null); // FIXME: Once ONOS-6977 is fixed |
| } |
| } |
| |
| // Activates all apps required by this application. |
| private void activateRequiredApps(Application app) { |
| app.requiredApps().stream().map(this::getId).forEach(id -> activate(id, app.id())); |
| } |
| |
| @Override |
| public void deactivate(ApplicationId appId) { |
| deactivateDependentApps(appId); |
| deactivate(appId, coreAppId); |
| } |
| |
| private void deactivate(ApplicationId appId, ApplicationId forAppId) { |
| requiredBy.remove(appId, forAppId); |
| if (requiredBy.get(appId).isEmpty()) { |
| AtomicBoolean stateChanged = new AtomicBoolean(false); |
| apps.computeIf(appId, |
| v -> v != null && v.state() != DEACTIVATED, |
| (k, v) -> { |
| stateChanged.set(true); |
| return new InternalApplicationHolder(v.app(), DEACTIVATED, v.permissions()); |
| }); |
| if (stateChanged.get()) { |
| updateTime(appId.name()); |
| deactivateRequiredApps(appId); |
| } |
| } |
| } |
| |
| // Deactivates all apps that require this application. |
| private void deactivateDependentApps(ApplicationId appId) { |
| apps.values() |
| .stream() |
| .map(Versioned::value) |
| .filter(a -> a.state() == ACTIVATED) |
| .filter(a -> a.app().requiredApps().contains(appId.name())) |
| .forEach(a -> deactivate(a.app().id())); |
| } |
| |
| // Deactivates all apps required by this application. |
| private void deactivateRequiredApps(ApplicationId appId) { |
| getApplication(appId).requiredApps() |
| .stream() |
| .map(this::getId) |
| .map(apps::get) |
| .map(Versioned::value) |
| .filter(a -> a.state() == ACTIVATED) |
| .forEach(a -> deactivate(a.app().id(), appId)); |
| } |
| |
| @Override |
| public Set<Permission> getPermissions(ApplicationId appId) { |
| InternalApplicationHolder app = Versioned.valueOrNull(apps.get(appId)); |
| return app != null ? ImmutableSet.copyOf(app.permissions()) : ImmutableSet.of(); |
| } |
| |
| @Override |
| public void setPermissions(ApplicationId appId, Set<Permission> permissions) { |
| AtomicBoolean permissionsChanged = new AtomicBoolean(false); |
| Versioned<InternalApplicationHolder> appHolder = apps.computeIf(appId, |
| v -> v != null && !Sets.symmetricDifference(v.permissions(), permissions).isEmpty(), |
| (k, v) -> { |
| permissionsChanged.set(true); |
| return new InternalApplicationHolder(v.app(), v.state(), ImmutableSet.copyOf(permissions)); |
| }); |
| if (permissionsChanged.get()) { |
| if (log.isTraceEnabled()) { |
| log.trace("Permission changed for {}", appId); |
| } |
| notifyDelegate(new ApplicationEvent(APP_PERMISSIONS_CHANGED, appHolder.value().app())); |
| } |
| } |
| |
| @Override |
| public InputStream getApplicationArchive(ApplicationId appId) { |
| return getApplicationInputStream(appId.name()); |
| } |
| |
| private class AppActivator implements Consumer<Application> { |
| @Override |
| public void accept(Application app) { |
| if (app != null) { // FIXME: Once ONOS-6977 is fixed |
| if (log.isTraceEnabled()) { |
| log.trace("Received an activation for {}", app.id()); |
| } |
| String appName = app.id().name(); |
| installAppIfNeeded(app); |
| setActive(appName); |
| boolean ready = localStartedApps.containsAll(app.requiredApps()); |
| if (ready && delegate != null) { |
| notifyDelegate(new ApplicationEvent(APP_ACTIVATED, app)); |
| localStartedApps.add(appName); |
| } else if (delegate == null) { |
| log.warn("Postponing app activation {} due to the delegate being null", app.id()); |
| } else { |
| log.warn("Postponing app activation {} due to req apps being not ready", app.id()); |
| } |
| } |
| } |
| } |
| |
| /** |
| * Listener to application state distributed map changes. |
| */ |
| private final class InternalAppsListener |
| implements MapEventListener<ApplicationId, InternalApplicationHolder> { |
| @Override |
| public void event(MapEvent<ApplicationId, InternalApplicationHolder> event) { |
| if (delegate == null) { |
| return; |
| } |
| |
| ApplicationId appId = event.key(); |
| InternalApplicationHolder newApp = event.newValue() == null ? null : event.newValue().value(); |
| InternalApplicationHolder oldApp = event.oldValue() == null ? null : event.oldValue().value(); |
| if (event.type() == MapEvent.Type.UPDATE && (newApp == null || oldApp == null || |
| newApp.state() == oldApp.state())) { |
| log.warn("Can't update the application {}", event.key()); |
| return; |
| } |
| if ((event.type() == MapEvent.Type.INSERT || event.type() == MapEvent.Type.UPDATE) && newApp != null) { |
| setupApplicationAndNotify(appId, newApp.app(), newApp.state()); |
| } else if (event.type() == MapEvent.Type.REMOVE && oldApp != null) { |
| if (log.isTraceEnabled()) { |
| log.trace("{} has been uninstalled", appId); |
| } |
| notifyDelegate(new ApplicationEvent(APP_UNINSTALLED, oldApp.app())); |
| purgeApplication(appId.name()); |
| localStartedApps.remove(appId.name()); |
| } else { |
| log.warn("Can't perform {} on application {}", event.type(), event.key()); |
| } |
| } |
| } |
| |
| private void setupApplicationAndNotify(ApplicationId appId, Application app, InternalState state) { |
| // ACTIVATED state is handled separately in NextAppToActivateValueListener |
| if (state == INSTALLED) { |
| fetchBitsIfNeeded(app); |
| if (log.isTraceEnabled()) { |
| log.trace("{} has been installed", app.id()); |
| } |
| notifyDelegate(new ApplicationEvent(APP_INSTALLED, app)); |
| } else if (state == DEACTIVATED) { |
| if (log.isTraceEnabled()) { |
| log.trace("{} has been deactivated", app.id()); |
| } |
| clearActive(appId.name()); |
| notifyDelegate(new ApplicationEvent(APP_DEACTIVATED, app)); |
| localStartedApps.remove(appId.name()); |
| } |
| } |
| |
| /** |
| * Determines if the application bits are available locally. |
| */ |
| private boolean appBitsAvailable(Application app) { |
| try { |
| ApplicationDescription appDesc = getApplicationDescription(app.id().name()); |
| return appDesc.version().equals(app.version()); |
| } catch (ApplicationException e) { |
| return false; |
| } |
| } |
| |
| /** |
| * Fetches the bits from the cluster peers if necessary. |
| */ |
| private void fetchBitsIfNeeded(Application app) { |
| if (!appBitsAvailable(app)) { |
| fetchBits(app, false); |
| } |
| } |
| |
| /** |
| * Installs the application if necessary from the application peers. |
| */ |
| private void installAppIfNeeded(Application app) { |
| if (!appBitsAvailable(app)) { |
| fetchBits(app, true); |
| } |
| } |
| |
| /** |
| * Fetches the bits from the cluster peers. |
| */ |
| private void fetchBits(Application app, boolean delegateInstallation) { |
| ControllerNode localNode = clusterService.getLocalNode(); |
| CountDownLatch latch = new CountDownLatch(1); |
| |
| // FIXME: send message with name & version to make sure we don't get served old bits |
| |
| log.info("Downloading bits for application {} for version : {}", app.id().name(), app.version()); |
| for (ControllerNode node : clusterService.getNodes()) { |
| if (latch.getCount() == 0) { |
| break; |
| } |
| if (node.equals(localNode)) { |
| continue; |
| } |
| clusterCommunicator.sendAndReceive(app.id().name(), |
| APP_BITS_REQUEST, |
| s -> s.getBytes(Charsets.UTF_8), |
| Function.identity(), |
| node.id()) |
| .whenCompleteAsync((bits, error) -> { |
| if (error == null && latch.getCount() > 0) { |
| saveApplication(new ByteArrayInputStream(bits)); |
| log.info("Downloaded bits for application {} from node {}", |
| app.id().name(), node.id()); |
| latch.countDown(); |
| if (delegateInstallation) { |
| if (log.isTraceEnabled()) { |
| log.trace("Delegate installation for {}", app.id()); |
| } |
| notifyDelegate(new ApplicationEvent(APP_INSTALLED, app)); |
| } |
| } else if (error != null) { |
| log.warn("Unable to fetch bits for application {} from node {}", |
| app.id().name(), node.id()); |
| } |
| }, messageHandlingExecutor); |
| } |
| |
| try { |
| if (!latch.await(FETCH_TIMEOUT_MS, MILLISECONDS)) { |
| log.warn("Unable to fetch bits for application {}", app.id().name()); |
| } |
| } catch (InterruptedException e) { |
| log.warn("Interrupted while fetching bits for application {}", app.id().name()); |
| Thread.currentThread().interrupt(); |
| } |
| } |
| |
| /** |
| * Produces a registered application from the supplied description. |
| */ |
| private Application registerApp(ApplicationDescription appDesc) { |
| ApplicationId appId = idStore.registerApplication(appDesc.name()); |
| return DefaultApplication |
| .builder(appDesc) |
| .withAppId(appId) |
| .build(); |
| } |
| |
| /** |
| * Internal class for holding app information. |
| */ |
| private static final class InternalApplicationHolder { |
| private final Application app; |
| private final InternalState state; |
| private final Set<Permission> permissions; |
| |
| @SuppressWarnings("unused") |
| private InternalApplicationHolder() { |
| app = null; |
| state = null; |
| permissions = null; |
| } |
| |
| private InternalApplicationHolder(Application app, InternalState state, Set<Permission> permissions) { |
| this.app = Preconditions.checkNotNull(app); |
| this.state = state; |
| this.permissions = permissions == null ? null : ImmutableSet.copyOf(permissions); |
| } |
| |
| public Application app() { |
| return app; |
| } |
| |
| public InternalState state() { |
| return state; |
| } |
| |
| public Set<Permission> permissions() { |
| return permissions; |
| } |
| |
| @Override |
| public String toString() { |
| return MoreObjects.toStringHelper(getClass()) |
| .add("app", app.id()) |
| .add("state", state) |
| .toString(); |
| } |
| } |
| } |