X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?p=controller.git;a=blobdiff_plain;f=opendaylight%2Fnetconf%2Fconfig-persister-impl%2Fsrc%2Fmain%2Fjava%2Forg%2Fopendaylight%2Fcontroller%2Fnetconf%2Fpersist%2Fimpl%2Fosgi%2FConfigPersisterActivator.java;h=6188492228aa55641e9770b839e0e04daf110263;hp=76afe8eb39b95f4500fd847cfe5df9a297c656e7;hb=5faa40e310344e006603456158931e1c4adf3251;hpb=2a4c88aa665a45c5394642cb3604603bebf8c0da diff --git a/opendaylight/netconf/config-persister-impl/src/main/java/org/opendaylight/controller/netconf/persist/impl/osgi/ConfigPersisterActivator.java b/opendaylight/netconf/config-persister-impl/src/main/java/org/opendaylight/controller/netconf/persist/impl/osgi/ConfigPersisterActivator.java index 76afe8eb39..6188492228 100644 --- a/opendaylight/netconf/config-persister-impl/src/main/java/org/opendaylight/controller/netconf/persist/impl/osgi/ConfigPersisterActivator.java +++ b/opendaylight/netconf/config-persister-impl/src/main/java/org/opendaylight/controller/netconf/persist/impl/osgi/ConfigPersisterActivator.java @@ -9,102 +9,90 @@ package org.opendaylight.controller.netconf.persist.impl.osgi; import com.google.common.annotations.VisibleForTesting; +import java.lang.management.ManagementFactory; +import java.util.ArrayList; +import java.util.List; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import javax.management.MBeanServer; +import org.opendaylight.controller.config.persist.api.ConfigPusher; import org.opendaylight.controller.config.persist.api.ConfigSnapshotHolder; +import org.opendaylight.controller.netconf.mapping.api.NetconfOperationProvider; import org.opendaylight.controller.netconf.mapping.api.NetconfOperationServiceFactory; -import org.opendaylight.controller.netconf.persist.impl.ConfigPersisterNotificationHandler; -import org.opendaylight.controller.netconf.persist.impl.ConfigPusher; +import org.opendaylight.controller.netconf.persist.impl.ConfigPusherImpl; import org.opendaylight.controller.netconf.persist.impl.PersisterAggregator; +import org.opendaylight.controller.netconf.util.CloseableUtil; import org.osgi.framework.BundleActivator; import org.osgi.framework.BundleContext; import org.osgi.framework.Constants; import org.osgi.framework.Filter; +import org.osgi.framework.InvalidSyntaxException; import org.osgi.framework.ServiceReference; +import org.osgi.framework.ServiceRegistration; import org.osgi.util.tracker.ServiceTracker; import org.osgi.util.tracker.ServiceTrackerCustomizer; import org.slf4j.Logger; import org.slf4j.LoggerFactory; -import javax.management.MBeanServer; -import java.lang.management.ManagementFactory; -import java.util.ArrayList; -import java.util.List; -import java.util.concurrent.TimeUnit; - public class ConfigPersisterActivator implements BundleActivator { - private static final Logger logger = LoggerFactory.getLogger(ConfigPersisterActivator.class); + private static final Logger LOG = LoggerFactory.getLogger(ConfigPersisterActivator.class); + private static final MBeanServer platformMBeanServer = ManagementFactory.getPlatformMBeanServer(); public static final String MAX_WAIT_FOR_CAPABILITIES_MILLIS_PROPERTY = "maxWaitForCapabilitiesMillis"; private static final long MAX_WAIT_FOR_CAPABILITIES_MILLIS_DEFAULT = TimeUnit.MINUTES.toMillis(2); public static final String CONFLICTING_VERSION_TIMEOUT_MILLIS_PROPERTY = "conflictingVersionTimeoutMillis"; - private static final long CONFLICTING_VERSION_TIMEOUT_MILLIS_DEFAULT = TimeUnit.SECONDS.toMillis(30); + private static final long CONFLICTING_VERSION_TIMEOUT_MILLIS_DEFAULT = TimeUnit.MINUTES.toMillis(1); public static final String NETCONF_CONFIG_PERSISTER = "netconf.config.persister"; public static final String STORAGE_ADAPTER_CLASS_PROP_SUFFIX = "storageAdapterClass"; - - private static final MBeanServer platformMBeanServer = ManagementFactory.getPlatformMBeanServer(); - private List autoCloseables; + private volatile BundleContext context; + ServiceRegistration registration; @Override public void start(final BundleContext context) throws Exception { - logger.debug("ConfigPersister starting"); + LOG.debug("ConfigPersister starting"); + this.context = context; + autoCloseables = new ArrayList<>(); PropertiesProviderBaseImpl propertiesProvider = new PropertiesProviderBaseImpl(context); - final PersisterAggregator persisterAggregator = PersisterAggregator.createFromProperties(propertiesProvider); autoCloseables.add(persisterAggregator); - final long maxWaitForCapabilitiesMillis = getMaxWaitForCapabilitiesMillis(propertiesProvider); - final List configs = persisterAggregator.loadLastConfigs(); - final long conflictingVersionTimeoutMillis = getConflictingVersionTimeoutMillis(propertiesProvider); - logger.trace("Following configs will be pushed: {}", configs); - ServiceTrackerCustomizer configNetconfCustomizer = new ServiceTrackerCustomizer() { - @Override - public NetconfOperationServiceFactory addingService(ServiceReference reference) { - NetconfOperationServiceFactory service = reference.getBundle().getBundleContext().getService(reference); - final ConfigPusher configPusher = new ConfigPusher(service, maxWaitForCapabilitiesMillis, conflictingVersionTimeoutMillis); - logger.debug("Configuration Persister got %s", service); - final Thread pushingThread = new Thread(new Runnable() { - @Override - public void run() { - configPusher.pushConfigs(configs); - logger.info("Configuration Persister initialization completed."); - ConfigPersisterNotificationHandler jmxNotificationHandler = new ConfigPersisterNotificationHandler(platformMBeanServer, persisterAggregator); - synchronized (ConfigPersisterActivator.this) { - autoCloseables.add(jmxNotificationHandler); - } - } - }, "config-pusher"); - synchronized (ConfigPersisterActivator.this){ - autoCloseables.add(new AutoCloseable() { - @Override - public void close() throws Exception { - pushingThread.interrupt(); - } - }); - } - pushingThread.start(); - return service; - } - - @Override - public void modifiedService(ServiceReference reference, NetconfOperationServiceFactory service) { - } + long maxWaitForCapabilitiesMillis = getMaxWaitForCapabilitiesMillis(propertiesProvider); + List configs = persisterAggregator.loadLastConfigs(); + long conflictingVersionTimeoutMillis = getConflictingVersionTimeoutMillis(propertiesProvider); + LOG.debug("Following configs will be pushed: {}", configs); + + InnerCustomizer innerCustomizer = new InnerCustomizer(configs, maxWaitForCapabilitiesMillis, + conflictingVersionTimeoutMillis, persisterAggregator); + OuterCustomizer outerCustomizer = new OuterCustomizer(context, innerCustomizer); + new ServiceTracker<>(context, NetconfOperationProvider.class, outerCustomizer).open(); + } - @Override - public void removedService(ServiceReference reference, NetconfOperationServiceFactory service) { - } - }; + private long getConflictingVersionTimeoutMillis(PropertiesProviderBaseImpl propertiesProvider) { + String timeoutProperty = propertiesProvider.getProperty(CONFLICTING_VERSION_TIMEOUT_MILLIS_PROPERTY); + return timeoutProperty == null ? CONFLICTING_VERSION_TIMEOUT_MILLIS_DEFAULT : Long.valueOf(timeoutProperty); + } - Filter filter = context.createFilter(getFilterString()); + private long getMaxWaitForCapabilitiesMillis(PropertiesProviderBaseImpl propertiesProvider) { + String timeoutProperty = propertiesProvider.getProperty(MAX_WAIT_FOR_CAPABILITIES_MILLIS_PROPERTY); + return timeoutProperty == null ? MAX_WAIT_FOR_CAPABILITIES_MILLIS_DEFAULT : Long.valueOf(timeoutProperty); + } - ServiceTracker tracker = - new ServiceTracker<>(context, filter, configNetconfCustomizer); - tracker.open(); + @Override + public void stop(BundleContext context) throws Exception { + synchronized(autoCloseables) { + CloseableUtil.closeAll(autoCloseables); + if (registration != null) { + registration.unregister(); + } + this.context = null; + } } @@ -116,32 +104,113 @@ public class ConfigPersisterActivator implements BundleActivator { ")"; } - private long getConflictingVersionTimeoutMillis(PropertiesProviderBaseImpl propertiesProvider) { - String timeoutProperty = propertiesProvider.getProperty(CONFLICTING_VERSION_TIMEOUT_MILLIS_PROPERTY); - return timeoutProperty == null ? CONFLICTING_VERSION_TIMEOUT_MILLIS_DEFAULT : Long.valueOf(timeoutProperty); - } + class OuterCustomizer implements ServiceTrackerCustomizer { + private final BundleContext context; + private final InnerCustomizer innerCustomizer; - private long getMaxWaitForCapabilitiesMillis(PropertiesProviderBaseImpl propertiesProvider) { - String timeoutProperty = propertiesProvider.getProperty(MAX_WAIT_FOR_CAPABILITIES_MILLIS_PROPERTY); - return timeoutProperty == null ? MAX_WAIT_FOR_CAPABILITIES_MILLIS_DEFAULT : Long.valueOf(timeoutProperty); - } + OuterCustomizer(BundleContext context, InnerCustomizer innerCustomizer) { + this.context = context; + this.innerCustomizer = innerCustomizer; + } - @Override - public synchronized void stop(BundleContext context) throws Exception { - Exception lastException = null; - for (AutoCloseable autoCloseable : autoCloseables) { + @Override + public NetconfOperationProvider addingService(ServiceReference reference) { + LOG.trace("Got OuterCustomizer.addingService {}", reference); + // JMX was registered, track config-netconf-connector + Filter filter; try { - autoCloseable.close(); - } catch (Exception e) { - if (lastException == null) { - lastException = e; - } else { - lastException.addSuppressed(e); + filter = context.createFilter(getFilterString()); + } catch (InvalidSyntaxException e) { + throw new IllegalStateException(e); + } + new ServiceTracker<>(context, filter, innerCustomizer).open(); + return null; + } + + @Override + public void modifiedService(ServiceReference reference, NetconfOperationProvider service) { + + } + + @Override + public void removedService(ServiceReference reference, NetconfOperationProvider service) { + + } + } + + class InnerCustomizer implements ServiceTrackerCustomizer { + private final List configs; + private final PersisterAggregator persisterAggregator; + private final long maxWaitForCapabilitiesMillis, conflictingVersionTimeoutMillis; + // This inner customizer has its filter to find the right operation service, but it gets triggered after any + // operation service appears. This means that it could start pushing thread up to N times (N = number of operation services spawned in OSGi) + private final AtomicBoolean alreadyStarted = new AtomicBoolean(false); + + InnerCustomizer(List configs, long maxWaitForCapabilitiesMillis, long conflictingVersionTimeoutMillis, + PersisterAggregator persisterAggregator) { + this.configs = configs; + this.maxWaitForCapabilitiesMillis = maxWaitForCapabilitiesMillis; + this.conflictingVersionTimeoutMillis = conflictingVersionTimeoutMillis; + this.persisterAggregator = persisterAggregator; + } + + @Override + public NetconfOperationServiceFactory addingService(ServiceReference reference) { + if(alreadyStarted.compareAndSet(false, true) == false) { + //Prevents multiple calls to this method spawning multiple pushing threads + return reference.getBundle().getBundleContext().getService(reference); + } + LOG.trace("Got InnerCustomizer.addingService {}", reference); + NetconfOperationServiceFactory service = reference.getBundle().getBundleContext().getService(reference); + + LOG.debug("Creating new job queue"); + + final ConfigPusherImpl configPusher = new ConfigPusherImpl(service, maxWaitForCapabilitiesMillis, conflictingVersionTimeoutMillis); + LOG.debug("Configuration Persister got {}", service); + LOG.debug("Context was {}", context); + LOG.debug("Registration was {}", registration); + + final Thread pushingThread = new Thread(new Runnable() { + @Override + public void run() { + try { + if(configs != null && !configs.isEmpty()) { + configPusher.pushConfigs(configs); + } + if(context != null) { + registration = context.registerService(ConfigPusher.class.getName(), configPusher, null); + configPusher.process(autoCloseables, platformMBeanServer, persisterAggregator); + } else { + LOG.warn("Unable to process configs as BundleContext is null"); + } + } catch (InterruptedException e) { + LOG.info("ConfigPusher thread stopped",e); + } + LOG.info("Configuration Persister initialization completed."); } + }, "config-pusher"); + synchronized (autoCloseables) { + autoCloseables.add(new AutoCloseable() { + @Override + public void close() { + pushingThread.interrupt(); + } + }); } + pushingThread.start(); + return service; + } + + @Override + public void modifiedService(ServiceReference reference, NetconfOperationServiceFactory service) { + LOG.trace("Got InnerCustomizer.modifiedService {}", reference); } - if (lastException != null) { - throw lastException; + + @Override + public void removedService(ServiceReference reference, NetconfOperationServiceFactory service) { + LOG.trace("Got InnerCustomizer.removedService {}", reference); } + } } +