package org.opendaylight.controller.blueprint;
import com.google.common.base.Optional;
+import com.google.common.base.Preconditions;
import com.google.common.collect.Lists;
import com.google.common.util.concurrent.ThreadFactoryBuilder;
import java.lang.management.ManagementFactory;
import java.util.AbstractMap.SimpleEntry;
+import java.util.ArrayDeque;
import java.util.ArrayList;
-import java.util.Dictionary;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.Deque;
import java.util.Hashtable;
import java.util.LinkedHashSet;
import java.util.List;
import javax.management.ObjectName;
import javax.xml.parsers.ParserConfigurationException;
import org.apache.aries.blueprint.services.BlueprintExtenderService;
+import org.apache.aries.quiesce.participant.QuiesceParticipant;
import org.apache.aries.util.AriesFrameworkUtil;
import org.opendaylight.controller.config.api.ConfigRegistry;
import org.opendaylight.controller.config.api.ConflictingVersionException;
import org.osgi.framework.BundleContext;
import org.osgi.framework.ServiceReference;
import org.osgi.framework.ServiceRegistration;
-import org.osgi.service.blueprint.container.EventConstants;
-import org.osgi.service.event.EventHandler;
+import org.osgi.service.blueprint.container.BlueprintEvent;
+import org.osgi.service.blueprint.container.BlueprintListener;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.w3c.dom.Document;
*/
class BlueprintContainerRestartServiceImpl implements AutoCloseable, BlueprintContainerRestartService {
private static final Logger LOG = LoggerFactory.getLogger(BlueprintContainerRestartServiceImpl.class);
+ private static final int CONTAINER_CREATE_TIMEOUT_IN_MINUTES = 5;
private static final String CONFIG_MODULE_NAMESPACE_PROP = "config-module-namespace";
private static final String CONFIG_MODULE_NAME_PROP = "config-module-name";
private static final String CONFIG_INSTANCE_NAME_PROP = "config-instance-name";
private final ExecutorService restartExecutor = Executors.newSingleThreadExecutor(new ThreadFactoryBuilder()
.setDaemon(true).setNameFormat("BlueprintContainerRestartService").build());
- private final BlueprintExtenderService blueprintExtenderService;
- BlueprintContainerRestartServiceImpl(BlueprintExtenderService blueprintExtenderService) {
+ private BlueprintExtenderService blueprintExtenderService;
+ private QuiesceParticipant quiesceParticipant;
+
+ void setBlueprintExtenderService(final BlueprintExtenderService blueprintExtenderService) {
this.blueprintExtenderService = blueprintExtenderService;
}
+ void setQuiesceParticipant(final QuiesceParticipant quiesceParticipant) {
+ this.quiesceParticipant = quiesceParticipant;
+ }
+
public void restartContainer(final Bundle bundle, final List<Object> paths) {
+ LOG.debug("restartContainer for bundle {}", bundle);
+
if (restartExecutor.isShutdown()) {
+ LOG.debug("Already closed - returning");
return;
}
- LOG.debug("restartContainer for bundle {}", bundle);
-
restartExecutor.execute(() -> {
blueprintExtenderService.destroyContainer(bundle, blueprintExtenderService.getContainer(bundle));
blueprintExtenderService.createContainer(bundle, paths);
restartExecutor.execute(() -> restartContainerAndDependentsInternal(bundle));
}
- private void restartContainerAndDependentsInternal(Bundle forBundle) {
+ private void restartContainerAndDependentsInternal(final Bundle forBundle) {
+ Preconditions.checkNotNull(blueprintExtenderService);
+ Preconditions.checkNotNull(quiesceParticipant);
+
// We use a LinkedHashSet to preserve insertion order as we walk the service usage hierarchy.
Set<Bundle> containerBundlesSet = new LinkedHashSet<>();
List<Entry<String, ModuleIdentifier>> configModules = new ArrayList<>();
LOG.info("Restarting blueprint containers for bundle {} and its dependent bundles {}", forBundle,
containerBundles.subList(1, containerBundles.size()));
- // Destroy the containers in reverse order with 'forBundle' last, ie bottom-up in the service tree.
- for (Bundle bundle: Lists.reverse(containerBundles)) {
- blueprintExtenderService.destroyContainer(bundle, blueprintExtenderService.getContainer(bundle));
- }
-
// The blueprint containers are created asynchronously so we register a handler for blueprint events
// that are sent when a container is complete, successful or not. The CountDownLatch tells when all
// containers are complete. This is done to ensure all blueprint containers are finished before we
// restart config modules.
final CountDownLatch containerCreationComplete = new CountDownLatch(containerBundles.size());
ServiceRegistration<?> eventHandlerReg = registerEventHandler(forBundle.getBundleContext(), event -> {
- LOG.debug("handleEvent {} for bundle {}", event.getTopic(), event.getProperty(EventConstants.BUNDLE));
- if (containerBundles.contains(event.getProperty(EventConstants.BUNDLE))) {
+ final Bundle bundle = event.getBundle();
+ if (event.isReplay()) {
+ LOG.trace("Got replay BlueprintEvent {} for bundle {}", event.getType(), bundle);
+ return;
+ }
+
+ LOG.debug("Got BlueprintEvent {} for bundle {}", event.getType(), bundle);
+ if (containerBundles.contains(bundle)
+ && (event.getType() == BlueprintEvent.CREATED || event.getType() == BlueprintEvent.FAILURE)) {
containerCreationComplete.countDown();
+ LOG.debug("containerCreationComplete is now {}", containerCreationComplete.getCount());
}
});
- // Restart the containers top-down starting with 'forBundle'.
- for (Bundle bundle: containerBundles) {
- List<Object> paths = BlueprintBundleTracker.findBlueprintPaths(bundle);
+ final Runnable createContainerCallback = () -> createContainers(containerBundles);
- LOG.info("Restarting blueprint container for bundle {} with paths {}", bundle, paths);
+ // Destroy the container down-top recursively and once done, restart the container top-down
+ destroyContainers(new ArrayDeque<>(Lists.reverse(containerBundles)), createContainerCallback);
- blueprintExtenderService.createContainer(bundle, paths);
- }
try {
- containerCreationComplete.await(5, TimeUnit.MINUTES);
- } catch (InterruptedException e) {
+ if (!containerCreationComplete.await(CONTAINER_CREATE_TIMEOUT_IN_MINUTES, TimeUnit.MINUTES)) {
+ LOG.warn("Failed to restart all blueprint containers within {} minutes. Attempted to restart {} {} "
+ + "but only {} completed restart", CONTAINER_CREATE_TIMEOUT_IN_MINUTES, containerBundles.size(),
+ containerBundles, containerBundles.size() - containerCreationComplete.getCount());
+ return;
+ }
+ } catch (final InterruptedException e) {
LOG.debug("CountDownLatch await was interrupted - returning");
return;
}
// Now restart any associated config system Modules.
restartConfigModules(forBundle.getBundleContext(), configModules);
+
+ LOG.info("Finished restarting blueprint containers for bundle {} and its dependent bundles", forBundle);
}
- private void restartConfigModules(BundleContext bundleContext, List<Entry<String,
+ /**
+ * Recursively quiesce and destroy the bundles one by one in order to maintain synchronicity and ordering.
+ * @param remainingBundlesToDestroy the list of remaining bundles to destroy.
+ * @param createContainerCallback a {@link Runnable} to {@code run()} when the recursive function is completed.
+ */
+ private void destroyContainers(final Deque<Bundle> remainingBundlesToDestroy,
+ final Runnable createContainerCallback) {
+
+ final Bundle nextBundle;
+ synchronized (remainingBundlesToDestroy) {
+ if (remainingBundlesToDestroy.isEmpty()) {
+ LOG.debug("All blueprint containers were quiesced and destroyed");
+ createContainerCallback.run();
+ return;
+ }
+
+ nextBundle = remainingBundlesToDestroy.poll();
+ }
+
+ // The Quiesce capability is a like a soft-stop, clean-stop. In the case of the Blueprint extender, in flight
+ // service calls are allowed to finish; they're counted in and counted out, and no new calls are allowed. When
+ // there are no in flight service calls, the bundle is told to stop. The Blueprint bundle itself doesn't know
+ // this is happening which is a key design point. In the case of Blueprint, the extender ensures no new Entity
+ // Managers(EMs) are created. Then when all those EMs are closed the quiesce operation reports that it is
+ // finished.
+ // To properly restart the blueprint containers, first we have to quiesce the list of bundles, and once done, it
+ // is safe to destroy their BlueprintContainer, so no reference is retained.
+ //
+ // Mail - thread explaining Quiesce API:
+ // https://www.mail-archive.com/dev@aries.apache.org/msg08403.html
+
+ // Quiesced the bundle to unregister the associated BlueprintContainer
+ quiesceParticipant.quiesce(bundlesQuiesced -> {
+
+ // Destroy the container once Quiesced
+ Arrays.stream(bundlesQuiesced).forEach(quiescedBundle -> {
+ LOG.debug("Quiesced bundle {}", quiescedBundle);
+ blueprintExtenderService.destroyContainer(
+ quiescedBundle, blueprintExtenderService.getContainer(quiescedBundle));
+ });
+
+ destroyContainers(remainingBundlesToDestroy, createContainerCallback);
+
+ }, Collections.singletonList(nextBundle));
+ }
+
+ private void createContainers(final List<Bundle> containerBundles) {
+ containerBundles.forEach(bundle -> {
+ List<Object> paths = BlueprintBundleTracker.findBlueprintPaths(bundle);
+
+ LOG.info("Restarting blueprint container for bundle {} with paths {}", bundle, paths);
+
+ blueprintExtenderService.createContainer(bundle, paths);
+ });
+ }
+
+ private void restartConfigModules(final BundleContext bundleContext, final List<Entry<String,
ModuleIdentifier>> configModules) {
if (configModules.isEmpty()) {
return;
return;
}
- ConfigSubsystemFacade configFacade = configFacadeFactory.createFacade("BlueprintContainerRestartService");
- try {
+ try (ConfigSubsystemFacade configFacade = configFacadeFactory.createFacade(
+ "BlueprintContainerRestartService")) {
restartConfigModules(configModules, configFacade);
} catch (ParserConfigurationException | DocumentedException | ValidationException
| ConflictingVersionException e) {
LOG.error("Error restarting config modules", e);
} finally {
- configFacade.close();
bundleContext.ungetService(configFacadeFactoryRef);
}
}
- private void restartConfigModules(List<Entry<String, ModuleIdentifier>> configModules,
- ConfigSubsystemFacade configFacade) throws ParserConfigurationException, DocumentedException,
+ private void restartConfigModules(final List<Entry<String, ModuleIdentifier>> configModules,
+ final ConfigSubsystemFacade configFacade) throws ParserConfigurationException, DocumentedException,
ValidationException, ConflictingVersionException {
Document document = XmlUtil.newDocument();
Config configMapping = configFacade.getConfigMapping();
ConfigRegistry configRegistryClient = new ConfigRegistryJMXClient(ManagementFactory.getPlatformMBeanServer());
- for (Entry<String, ModuleIdentifier> entry: configModules) {
+ for (Entry<String, ModuleIdentifier> entry : configModules) {
String moduleNamespace = entry.getKey();
ModuleIdentifier moduleId = entry.getValue();
try {
Element moduleElement = configMapping.moduleToXml(moduleNamespace, moduleId.getFactoryName(),
moduleId.getInstanceName(), instanceON, document);
modulesElement.appendChild(moduleElement);
- } catch (InstanceNotFoundException e) {
+ } catch (final InstanceNotFoundException e) {
LOG.warn("Error looking up config module: namespace {}, module name {}, instance {}",
moduleNamespace, moduleId.getFactoryName(), moduleId.getInstanceName(), e);
}
*
* @param bundle the bundle to traverse
* @param containerBundles the current set of bundles containing blueprint containers
+ * @param configModules the current set of bundles containing config modules
*/
- private void findDependentContainersRecursively(Bundle bundle, Set<Bundle> containerBundles,
- List<Entry<String, ModuleIdentifier>> configModules) {
+ private void findDependentContainersRecursively(final Bundle bundle, final Set<Bundle> containerBundles,
+ final List<Entry<String, ModuleIdentifier>> configModules) {
if (!containerBundles.add(bundle)) {
// Already seen this bundle...
return;
}
}
- private void possiblyAddConfigModuleIdentifier(ServiceReference<?> reference,
- List<Entry<String, ModuleIdentifier>> configModules) {
+ private void possiblyAddConfigModuleIdentifier(final ServiceReference<?> reference,
+ final List<Entry<String, ModuleIdentifier>> configModules) {
Object moduleNamespace = reference.getProperty(CONFIG_MODULE_NAMESPACE_PROP);
if (moduleNamespace == null) {
return;
}
@Nullable
- private String getRequiredConfigModuleProperty(String propName, Object moduleNamespace,
- ServiceReference<?> reference) {
+ private String getRequiredConfigModuleProperty(final String propName, final Object moduleNamespace,
+ final ServiceReference<?> reference) {
Object value = reference.getProperty(propName);
if (value == null) {
LOG.warn(
return value.toString();
}
- private ServiceRegistration<?> registerEventHandler(BundleContext bundleContext, EventHandler handler) {
- Dictionary<String, Object> props = new Hashtable<>();
- props.put(org.osgi.service.event.EventConstants.EVENT_TOPIC,
- new String[]{EventConstants.TOPIC_CREATED, EventConstants.TOPIC_FAILURE});
- return bundleContext.registerService(EventHandler.class.getName(), handler, props);
+ private ServiceRegistration<?> registerEventHandler(final BundleContext bundleContext,
+ final BlueprintListener listener) {
+ return bundleContext.registerService(BlueprintListener.class.getName(), listener, new Hashtable<>());
}
@Override
public void close() {
+ LOG.debug("Closing");
+
restartExecutor.shutdownNow();
}
}
Code Review / controller.git / blobdiff